mirror of
https://github.com/esphome/esphome.git
synced 2026-10-12 17:23:04 +00:00
[motion] Add gesture detection (#16787)
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@@ -19,10 +20,20 @@ from esphome.components.motion import (
|
||||
_transform_matrix,
|
||||
_validate_matrix_options,
|
||||
)
|
||||
from esphome.components.motion.binary_sensor import (
|
||||
CONFIG_SCHEMA as BINARY_SENSOR_CONFIG_SCHEMA,
|
||||
to_code as binary_sensor_to_code,
|
||||
)
|
||||
from esphome.components.motion.event import (
|
||||
CONFIG_SCHEMA as EVENT_CONFIG_SCHEMA,
|
||||
to_code as event_to_code,
|
||||
)
|
||||
from esphome.components.motion.sensor import (
|
||||
_ACCELERATIONS,
|
||||
_ANGULAR_RATES,
|
||||
_GYROSCOPES,
|
||||
CONF_FLAT_THRESHOLD,
|
||||
CONF_ORIENTATION,
|
||||
CONF_PITCH,
|
||||
CONF_ROLL,
|
||||
CONFIG_SCHEMA,
|
||||
@@ -263,7 +274,7 @@ class TestAxisMapToMatrix:
|
||||
|
||||
def _expr_str(sensor_type: str) -> str:
|
||||
"""Build a sensor expression via the production function and return its string form."""
|
||||
return str(build_sensor_expr(sensor_type, MockObj("data")))
|
||||
return str(build_sensor_expr(sensor_type, MockObj("data"), {}))
|
||||
|
||||
|
||||
class TestSensorExpressions:
|
||||
@@ -326,6 +337,19 @@ class TestSensorExpressions:
|
||||
# Pitch negates the x component
|
||||
assert "(-data.acceleration[0])" in expr
|
||||
|
||||
def test_orientation_expression_custom_threshold(self):
|
||||
"""The configured flat_threshold (degrees) is converted to a sine and passed
|
||||
to the helper."""
|
||||
|
||||
expr = str(
|
||||
build_sensor_expr(
|
||||
CONF_ORIENTATION, MockObj("data"), {CONF_FLAT_THRESHOLD: 45}
|
||||
)
|
||||
)
|
||||
assert "orientation_degrees(data" in expr
|
||||
expected = round(math.sin(math.radians(45)), 6) # 0.707107
|
||||
assert str(expected) in expr
|
||||
|
||||
|
||||
# --- Calibration math ---
|
||||
#
|
||||
@@ -372,7 +396,6 @@ def _calibrate_level(
|
||||
|
||||
Composes the correction with *matrix* (defaults to identity).
|
||||
"""
|
||||
import math
|
||||
|
||||
if matrix is None:
|
||||
matrix = list(IDENTITY)
|
||||
@@ -410,7 +433,6 @@ def _calibrate_level(
|
||||
|
||||
def _calibrate_heading(matrix: list[float], raw: list[float]) -> list[float]:
|
||||
"""Python port of MotionComponent::calibrate_heading."""
|
||||
import math
|
||||
|
||||
mapped = _mat_vec(matrix, raw)
|
||||
mx, my = mapped[0], mapped[1]
|
||||
@@ -438,7 +460,6 @@ class TestCalibrateLevel:
|
||||
|
||||
def _assert_maps_to_z(self, raw: list[float]) -> list[float]:
|
||||
"""Assert that the calibration matrix maps raw to [0, 0, 1]."""
|
||||
import math
|
||||
|
||||
m = _calibrate_level(raw)
|
||||
mag = math.sqrt(sum(v * v for v in raw))
|
||||
@@ -461,7 +482,6 @@ class TestCalibrateLevel:
|
||||
|
||||
def test_composes_with_existing_matrix(self):
|
||||
"""Level calibration should correct tilt while preserving an existing axis swap."""
|
||||
import math
|
||||
|
||||
swap = [0, 1, 0, 1, 0, 0, 0, 0, 1] # swap X↔Y
|
||||
# Tilted raw: gravity has X component in raw frame
|
||||
@@ -495,7 +515,6 @@ class TestCalibrateLevel:
|
||||
self._assert_maps_to_z([0, 1.0, 0])
|
||||
|
||||
def test_tilted_45_degrees(self):
|
||||
import math
|
||||
|
||||
self._assert_maps_to_z(
|
||||
[math.sin(math.radians(45)), 0, math.cos(math.radians(45))]
|
||||
@@ -534,7 +553,6 @@ class TestCalibrateHeading:
|
||||
|
||||
def test_y_axis_tilt_no_heading_error(self):
|
||||
"""Device tilted purely around Y — heading should already be correct."""
|
||||
import math
|
||||
|
||||
flat_raw = [0, 0, 1.0]
|
||||
level_m = _calibrate_level(flat_raw)
|
||||
@@ -547,7 +565,6 @@ class TestCalibrateHeading:
|
||||
|
||||
def test_corrects_heading_rotation(self):
|
||||
"""After level+heading calibration, mapped Y should be ~0 when tilted."""
|
||||
import math
|
||||
|
||||
# Simulate a sensor whose chip is rotated 30° around Z relative to enclosure
|
||||
angle = math.radians(30)
|
||||
@@ -573,7 +590,6 @@ class TestCalibrateHeading:
|
||||
|
||||
def test_full_calibration_sequence(self):
|
||||
"""End-to-end: level then heading produces correct frame alignment."""
|
||||
import math
|
||||
|
||||
# Chip is mounted tilted 15° around Y and 25° around Z
|
||||
# Build the chip-to-enclosure rotation: Rz(25°) * Ry(15°)
|
||||
@@ -846,7 +862,10 @@ class TestSensorConfigSchema:
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"sensor_type",
|
||||
_ACCELERATIONS + _GYROSCOPES + _ANGULAR_RATES + [CONF_PITCH, CONF_ROLL],
|
||||
_ACCELERATIONS
|
||||
+ _GYROSCOPES
|
||||
+ _ANGULAR_RATES
|
||||
+ [CONF_PITCH, CONF_ROLL, CONF_ORIENTATION],
|
||||
)
|
||||
def test_valid_types_accepted(self, sensor_type):
|
||||
"""Valid sensor types should pass type validation (errors from missing
|
||||
@@ -858,3 +877,211 @@ class TestSensorConfigSchema:
|
||||
assert "Unknown value" not in str(e), (
|
||||
f"Type '{sensor_type}' was rejected as unknown"
|
||||
)
|
||||
|
||||
|
||||
# --- Binary sensor & Event platform tests ---
|
||||
|
||||
|
||||
class TestBinarySensorSchema:
|
||||
def test_valid_types(self):
|
||||
res = BINARY_SENSOR_CONFIG_SCHEMA(
|
||||
{
|
||||
"type": "face_up",
|
||||
"motion_id": "my_motion_component",
|
||||
"name": "LSM6DS3 Face Up",
|
||||
}
|
||||
)
|
||||
assert res["type"] == "face_up"
|
||||
assert str(res["motion_id"]) == "my_motion_component"
|
||||
# face_up / face_down configure their threshold as a tilt angle in degrees.
|
||||
assert res["threshold"] == pytest.approx(30.0)
|
||||
|
||||
def test_free_fall_defaults(self):
|
||||
res = BINARY_SENSOR_CONFIG_SCHEMA(
|
||||
{
|
||||
"type": "free_fall",
|
||||
"motion_id": "my_motion_component",
|
||||
"name": "LSM6DS3 Free Fall",
|
||||
}
|
||||
)
|
||||
assert res["threshold"] == pytest.approx(0.15)
|
||||
assert res["duration"].total_milliseconds == 100
|
||||
|
||||
@pytest.mark.parametrize("sensor_type", ["free_fall", "moving"])
|
||||
def test_zero_threshold_rejected(self, sensor_type: str) -> None:
|
||||
with pytest.raises((Invalid, MultipleInvalid)):
|
||||
BINARY_SENSOR_CONFIG_SCHEMA(
|
||||
{
|
||||
"type": sensor_type,
|
||||
"motion_id": "my_motion_component",
|
||||
"name": "Test",
|
||||
"threshold": 0,
|
||||
}
|
||||
)
|
||||
|
||||
def test_invalid_type(self):
|
||||
with pytest.raises((Invalid, MultipleInvalid)):
|
||||
BINARY_SENSOR_CONFIG_SCHEMA(
|
||||
{
|
||||
"type": "invalid_type",
|
||||
"motion_id": "my_motion_component",
|
||||
"name": "Test",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class TestEventSchema:
|
||||
def test_valid_event(self):
|
||||
res = EVENT_CONFIG_SCHEMA(
|
||||
{"motion_id": "my_motion_component", "name": "LSM6DS3 Shake"}
|
||||
)
|
||||
assert str(res["motion_id"]) == "my_motion_component"
|
||||
assert res["threshold"] == pytest.approx(0.5)
|
||||
assert res["cooldown"].total_milliseconds == 500
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_binary_sensor_codegen():
|
||||
mock_var = MagicMock()
|
||||
mock_parent = MagicMock()
|
||||
with (
|
||||
patch(
|
||||
"esphome.components.motion.binary_sensor.cg.get_variable",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_parent,
|
||||
) as mock_get_var,
|
||||
patch(
|
||||
"esphome.components.motion.binary_sensor.cg.new_Pvariable",
|
||||
return_value=mock_var,
|
||||
) as mock_new_pvar,
|
||||
patch(
|
||||
"esphome.components.motion.binary_sensor.binary_sensor.register_binary_sensor",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_reg_bin,
|
||||
patch(
|
||||
"esphome.components.motion.binary_sensor.cg.register_component",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_reg_comp,
|
||||
patch(
|
||||
"esphome.components.motion.binary_sensor.cg.add",
|
||||
) as mock_add,
|
||||
):
|
||||
yield {
|
||||
"get_variable": mock_get_var,
|
||||
"new_Pvariable": mock_new_pvar,
|
||||
"register_binary_sensor": mock_reg_bin,
|
||||
"register_component": mock_reg_comp,
|
||||
"add": mock_add,
|
||||
"var": mock_var,
|
||||
"parent": mock_parent,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_binary_sensor_to_code(mock_binary_sensor_codegen):
|
||||
|
||||
# face_up has no duration: only the threshold is configured.
|
||||
config = {
|
||||
"id": "my_binary_sensor_id",
|
||||
"type": "face_up",
|
||||
"motion_id": "my_motion_component",
|
||||
"threshold": 30.0,
|
||||
}
|
||||
await binary_sensor_to_code(config)
|
||||
mock_binary_sensor_codegen["get_variable"].assert_called_once_with(
|
||||
"my_motion_component"
|
||||
)
|
||||
mock_binary_sensor_codegen["new_Pvariable"].assert_called_once()
|
||||
mock_binary_sensor_codegen["register_binary_sensor"].assert_called_once_with(
|
||||
mock_binary_sensor_codegen["var"], config
|
||||
)
|
||||
mock_binary_sensor_codegen["register_component"].assert_called_once_with(
|
||||
mock_binary_sensor_codegen["var"], config
|
||||
)
|
||||
assert mock_binary_sensor_codegen["add"].call_count == 1
|
||||
# face_up threshold is configured in degrees and converted to a cosine for C++.
|
||||
mock_binary_sensor_codegen["var"].set_threshold.assert_called_once_with(
|
||||
pytest.approx(math.cos(math.radians(30.0)))
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_binary_sensor_to_code_with_duration(mock_binary_sensor_codegen):
|
||||
from esphome.core import TimePeriod
|
||||
|
||||
# free_fall has a duration: both threshold and duration are configured, and the
|
||||
# threshold is passed straight through (no degrees-to-cosine conversion).
|
||||
config = {
|
||||
"id": "my_binary_sensor_id",
|
||||
"type": "free_fall",
|
||||
"motion_id": "my_motion_component",
|
||||
"threshold": 0.15,
|
||||
"duration": TimePeriod(milliseconds=100),
|
||||
}
|
||||
await binary_sensor_to_code(config)
|
||||
assert mock_binary_sensor_codegen["add"].call_count == 2
|
||||
mock_binary_sensor_codegen["var"].set_threshold.assert_called_once_with(0.15)
|
||||
mock_binary_sensor_codegen["var"].set_duration.assert_called_once_with(
|
||||
TimePeriod(milliseconds=100)
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_event_codegen():
|
||||
mock_var = MagicMock()
|
||||
mock_parent = MagicMock()
|
||||
with (
|
||||
patch(
|
||||
"esphome.components.motion.event.cg.get_variable",
|
||||
new_callable=AsyncMock,
|
||||
return_value=mock_parent,
|
||||
) as mock_get_var,
|
||||
patch(
|
||||
"esphome.components.motion.event.cg.new_Pvariable",
|
||||
return_value=mock_var,
|
||||
) as mock_new_pvar,
|
||||
patch(
|
||||
"esphome.components.motion.event.event.register_event",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_reg_event,
|
||||
patch(
|
||||
"esphome.components.motion.event.cg.register_component",
|
||||
new_callable=AsyncMock,
|
||||
) as mock_reg_comp,
|
||||
patch(
|
||||
"esphome.components.motion.event.cg.add",
|
||||
) as mock_add,
|
||||
):
|
||||
yield {
|
||||
"get_variable": mock_get_var,
|
||||
"new_Pvariable": mock_new_pvar,
|
||||
"register_event": mock_reg_event,
|
||||
"register_component": mock_reg_comp,
|
||||
"add": mock_add,
|
||||
"var": mock_var,
|
||||
"parent": mock_parent,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_event_to_code(mock_event_codegen):
|
||||
from esphome.core import TimePeriod
|
||||
|
||||
config = {
|
||||
"id": "my_event_id",
|
||||
"motion_id": "my_motion_component",
|
||||
"threshold": 0.5,
|
||||
"cooldown": TimePeriod(milliseconds=500),
|
||||
}
|
||||
await event_to_code(config)
|
||||
mock_event_codegen["get_variable"].assert_called_once_with("my_motion_component")
|
||||
mock_event_codegen["new_Pvariable"].assert_called_once_with(
|
||||
"my_event_id", mock_event_codegen["parent"]
|
||||
)
|
||||
mock_event_codegen["register_event"].assert_called_once_with(
|
||||
mock_event_codegen["var"], config, event_types=["shake"]
|
||||
)
|
||||
mock_event_codegen["register_component"].assert_called_once_with(
|
||||
mock_event_codegen["var"], config
|
||||
)
|
||||
assert mock_event_codegen["add"].call_count == 2
|
||||
|
||||
@@ -57,11 +57,40 @@ sensor:
|
||||
motion_id: qmi8658_motion
|
||||
type: roll
|
||||
name: "Roll"
|
||||
- platform: motion
|
||||
motion_id: qmi8658_motion
|
||||
type: orientation
|
||||
name: "Orientation"
|
||||
flat_threshold: 30
|
||||
|
||||
binary_sensor:
|
||||
- platform: motion
|
||||
motion_id: qmi8658_motion
|
||||
type: face_up
|
||||
name: "Face Up"
|
||||
- platform: motion
|
||||
motion_id: qmi8658_motion
|
||||
type: face_down
|
||||
name: "Face Down"
|
||||
- platform: motion
|
||||
motion_id: qmi8658_motion
|
||||
type: free_fall
|
||||
name: "Free Fall"
|
||||
- platform: motion
|
||||
motion_id: qmi8658_motion
|
||||
type: moving
|
||||
name: "Moving"
|
||||
|
||||
event:
|
||||
- platform: motion
|
||||
motion_id: qmi8658_motion
|
||||
name: "Shake"
|
||||
|
||||
motion:
|
||||
- platform: qmi8658
|
||||
id: qmi8658_motion
|
||||
i2c_id: i2c_bus
|
||||
update_interval: 100ms
|
||||
# Accelerometer full-scale range: 2G | 4G | 8G | 16G
|
||||
accelerometer_range: 4G
|
||||
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Tests for the motion platforms' update_interval warning (shake, free_fall, moving)."""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome import config, yaml_util
|
||||
from esphome.core import CORE
|
||||
|
||||
CONFIG_TEMPLATE = """
|
||||
esphome:
|
||||
name: test
|
||||
|
||||
esp8266:
|
||||
board: esp01_1m
|
||||
|
||||
i2c:
|
||||
sda: GPIO4
|
||||
scl: GPIO5
|
||||
|
||||
motion:
|
||||
- platform: lsm6ds
|
||||
id: lsm6ds_motion
|
||||
update_interval: {update_interval}
|
||||
|
||||
{platform_config}
|
||||
"""
|
||||
|
||||
EVENT_CONFIG = """
|
||||
event:
|
||||
- platform: motion
|
||||
motion_id: lsm6ds_motion
|
||||
name: "Shake event"
|
||||
"""
|
||||
|
||||
FREE_FALL_CONFIG = """
|
||||
binary_sensor:
|
||||
- platform: motion
|
||||
motion_id: lsm6ds_motion
|
||||
type: free_fall
|
||||
name: "Free fall"
|
||||
"""
|
||||
|
||||
MOVING_CONFIG = """
|
||||
binary_sensor:
|
||||
- platform: motion
|
||||
motion_id: lsm6ds_motion
|
||||
type: moving
|
||||
name: "Moving"
|
||||
"""
|
||||
|
||||
FACE_UP_CONFIG = """
|
||||
binary_sensor:
|
||||
- platform: motion
|
||||
motion_id: lsm6ds_motion
|
||||
type: face_up
|
||||
name: "Face up"
|
||||
"""
|
||||
|
||||
|
||||
def _read_config(tmp_path: Path, update_interval: str, platform_config: str):
|
||||
test_file = tmp_path / "test.yaml"
|
||||
test_file.write_text(
|
||||
CONFIG_TEMPLATE.format(
|
||||
update_interval=update_interval, platform_config=platform_config
|
||||
)
|
||||
)
|
||||
|
||||
parsed_yaml = yaml_util.load_yaml(test_file)
|
||||
|
||||
with (
|
||||
patch.object(yaml_util, "load_yaml", return_value=parsed_yaml),
|
||||
patch.object(CORE, "config_path", test_file),
|
||||
):
|
||||
return config.read_config({})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("platform_config", "feature_name"),
|
||||
[
|
||||
(EVENT_CONFIG, "shake"),
|
||||
(FREE_FALL_CONFIG, "free-fall"),
|
||||
(MOVING_CONFIG, "moving"),
|
||||
],
|
||||
ids=["shake", "free_fall", "moving"],
|
||||
)
|
||||
def test_warns_on_slow_update_interval(
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
platform_config: str,
|
||||
feature_name: str,
|
||||
) -> None:
|
||||
"""A parent update_interval slower than 100ms should log a warning."""
|
||||
with caplog.at_level(logging.WARNING):
|
||||
result = _read_config(tmp_path, "250ms", platform_config)
|
||||
|
||||
assert result is not None, "Slow update_interval should still be a valid config"
|
||||
warning_text = (
|
||||
f"{feature_name} detection works best with an update_interval of 100ms or less"
|
||||
)
|
||||
assert any(warning_text in record.message for record in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("platform_config", "feature_name"),
|
||||
[
|
||||
(EVENT_CONFIG, "shake"),
|
||||
(FREE_FALL_CONFIG, "free-fall"),
|
||||
(MOVING_CONFIG, "moving"),
|
||||
],
|
||||
ids=["shake", "free_fall", "moving"],
|
||||
)
|
||||
def test_no_warning_on_fast_update_interval(
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
platform_config: str,
|
||||
feature_name: str,
|
||||
) -> None:
|
||||
"""A parent update_interval of 100ms or less should not log a warning."""
|
||||
with caplog.at_level(logging.WARNING):
|
||||
result = _read_config(tmp_path, "50ms", platform_config)
|
||||
|
||||
assert result is not None
|
||||
warning_text = f"{feature_name} detection works best with an update_interval"
|
||||
assert not any(warning_text in record.message for record in caplog.records)
|
||||
|
||||
|
||||
def test_no_warning_for_face_up_with_slow_update_interval(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""face_up/face_down track steady orientation, not a fast event, so no warning."""
|
||||
with caplog.at_level(logging.WARNING):
|
||||
result = _read_config(tmp_path, "250ms", FACE_UP_CONFIG)
|
||||
|
||||
assert result is not None
|
||||
assert not any(
|
||||
"detection works best with an update_interval" in record.message
|
||||
for record in caplog.records
|
||||
)
|
||||
Reference in New Issue
Block a user