jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Accept more than 1 state for numeric entities in Bayesian (#119281)

Browse Source 
* test driven delevopment

* test driven development - multi numeric state

* better multi-state processing

* when state==below return true

* adds test for a bad state

* improve codecov

* value error already handled in async_numeric_state

* remove whitespace

* remove async_get

* linting

* test_driven dev for error handling

* make tests fail correctly

* ensure tests fail correctly

* prevent bad numeric entries

* ensure no overlapping ranges

* fix tests, as error caught in validation

* remove redundant er call

* remove reddundant arg

* improves code coverage

* filter for numeric states before testing overlap

* adress code review

* skip non numeric configs but continue

* wait to avoid race condition

* Better tuples name and better guard clause

* better test description

* more accurate description

* Add comments to calculations

* using typing not collections as per ruff

* Apply suggestions from code review

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* follow on from suggestions

* Lazy evaluation

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* update error text in tests

* fix broken tests

* move validation function call

* fixes return type of above_greater_than_below.

* improves codecov

* fixes validation

---------

Co-authored-by: Erik Montnemery <erik@montnemery.com>
This commit is contained in:
HarvsG 2024-09-12 11:06:18 +01:00 committed by GitHub
parent c21ea6b8da
commit 70ebf2f5d8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 464 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

149
homeassistant/components/bayesian/binary_sensor.py
View File

@ -5,7 +5,8 @@ from __future__ import annotations
from collections import OrderedDict
from collections.abc import Callable
import logging
from typing import Any
import math
from typing import TYPE_CHECKING, Any, NamedTuple
from uuid import UUID
import voluptuous as vol
@ -50,6 +51,7 @@ from .const import (
ATTR_OCCURRED_OBSERVATION_ENTITIES,
ATTR_PROBABILITY,
ATTR_PROBABILITY_THRESHOLD,
CONF_NUMERIC_STATE,
CONF_OBSERVATIONS,
CONF_P_GIVEN_F,
CONF_P_GIVEN_T,
@ -66,18 +68,74 @@ from .issues import raise_mirrored_entries, raise_no_prob_given_false
_LOGGER = logging.getLogger(__name__)
NUMERIC_STATE_SCHEMA = vol.Schema(
{
CONF_PLATFORM: "numeric_state",
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Optional(CONF_ABOVE): vol.Coerce(float),
vol.Optional(CONF_BELOW): vol.Coerce(float),
vol.Required(CONF_P_GIVEN_T): vol.Coerce(float),
vol.Optional(CONF_P_GIVEN_F): vol.Coerce(float),
},
required=True,
def _above_greater_than_below(config: dict[str, Any]) -> dict[str, Any]:
if config[CONF_PLATFORM] == CONF_NUMERIC_STATE:
above = config.get(CONF_ABOVE)
below = config.get(CONF_BELOW)
if above is None and below is None:
_LOGGER.error(
"For bayesian numeric state for entity: %s at least one of 'above' or 'below' must be specified",
config[CONF_ENTITY_ID],
)
raise vol.Invalid(
"For bayesian numeric state at least one of 'above' or 'below' must be specified."
)
if above is not None and below is not None:
if above > below:
_LOGGER.error(
"For bayesian numeric state 'above' (%s) must be less than 'below' (%s)",
above,
below,
)
raise vol.Invalid("'above' is greater than 'below'")
return config
NUMERIC_STATE_SCHEMA = vol.All(
vol.Schema(
{
CONF_PLATFORM: CONF_NUMERIC_STATE,
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Optional(CONF_ABOVE): vol.Coerce(float),
vol.Optional(CONF_BELOW): vol.Coerce(float),
vol.Required(CONF_P_GIVEN_T): vol.Coerce(float),
vol.Optional(CONF_P_GIVEN_F): vol.Coerce(float),
},
required=True,
),
_above_greater_than_below,
)
def _no_overlapping(configs: list[dict]) -> list[dict]:
numeric_configs = [
config for config in configs if config[CONF_PLATFORM] == CONF_NUMERIC_STATE
]
if len(numeric_configs) < 2:
return configs
class NumericConfig(NamedTuple):
above: float
below: float
d: dict[str, list[NumericConfig]] = {}
for _, config in enumerate(numeric_configs):
above = config.get(CONF_ABOVE, -math.inf)
below = config.get(CONF_BELOW, math.inf)
entity_id: str = str(config[CONF_ENTITY_ID])
d.setdefault(entity_id, []).append(NumericConfig(above, below))
for ent_id, intervals in d.items():
intervals = sorted(intervals, key=lambda tup: tup.above)
for i, tup in enumerate(intervals):
if len(intervals) > i + 1 and tup.below > intervals[i + 1].above:
raise vol.Invalid(
f"Ranges for bayesian numeric state entities must not overlap, but {ent_id} has overlapping ranges, above:{tup.above}, below:{tup.below} overlaps with above:{intervals[i+1].above}, below:{intervals[i+1].below}."
)
return configs
STATE_SCHEMA = vol.Schema(
{
CONF_PLATFORM: CONF_STATE,
@ -107,7 +165,8 @@ PLATFORM_SCHEMA = BINARY_SENSOR_PLATFORM_SCHEMA.extend(
vol.Required(CONF_OBSERVATIONS): vol.Schema(
vol.All(
cv.ensure_list,
[vol.Any(NUMERIC_STATE_SCHEMA, STATE_SCHEMA, TEMPLATE_SCHEMA)],
[vol.Any(TEMPLATE_SCHEMA, STATE_SCHEMA, NUMERIC_STATE_SCHEMA)],
_no_overlapping,
)
),
vol.Required(CONF_PRIOR): vol.Coerce(float),
@ -211,10 +270,11 @@ class BayesianBinarySensor(BinarySensorEntity):
self.observations_by_entity = self._build_observations_by_entity()
self.observations_by_template = self._build_observations_by_template()
self.observation_handlers: dict[str, Callable[[Observation], bool | None]] = {
self.observation_handlers: dict[
str, Callable[[Observation, bool], bool | None]
] = {
"numeric_state": self._process_numeric_state,
"state": self._process_state,
"multi_state": self._process_multi_state,
}
async def async_added_to_hass(self) -> None:
@ -342,8 +402,9 @@ class BayesianBinarySensor(BinarySensorEntity):
for observation in self.observations_by_entity[entity]:
platform = observation.platform
observation.observed = self.observation_handlers[platform](observation)
observation.observed = self.observation_handlers[platform](
observation, observation.multi
)
local_observations[observation.id] = observation
return local_observations
@ -408,9 +469,7 @@ class BayesianBinarySensor(BinarySensorEntity):
if len(entity_observations) == 1:
continue
for observation in entity_observations:
if observation.platform != "state":
continue
observation.platform = "multi_state"
observation.multi = True
return observations_by_entity
@ -437,14 +496,23 @@ class BayesianBinarySensor(BinarySensorEntity):
return observations_by_template
def _process_numeric_state(self, entity_observation: Observation) -> bool | None:
def _process_numeric_state(
self, entity_observation: Observation, multi: bool = False
) -> bool | None:
"""Return True if numeric condition is met, return False if not, return None otherwise."""
entity = entity_observation.entity_id
entity_id = entity_observation.entity_id
# if we are dealing with numeric_state observations entity_id cannot be None
if TYPE_CHECKING:
assert entity_id is not None
entity = self.hass.states.get(entity_id)
if entity is None:
return None
try:
if condition.state(self.hass, entity, [STATE_UNKNOWN, STATE_UNAVAILABLE]):
return None
return condition.async_numeric_state(
result = condition.async_numeric_state(
self.hass,
entity,
entity_observation.below,
@ -452,10 +520,24 @@ class BayesianBinarySensor(BinarySensorEntity):
None,
entity_observation.to_dict(),
)
if result:
return True
if multi:
state = float(entity.state)
if (
entity_observation.below is not None
and state == entity_observation.below
):
return True
return None
except ConditionError:
return None
else:
return False
def _process_state(self, entity_observation: Observation) -> bool | None:
def _process_state(
self, entity_observation: Observation, multi: bool = False
) -> bool | None:
"""Return True if state conditions are met, return False if they are not.
Returns None if the state is unavailable.
@ -467,24 +549,13 @@ class BayesianBinarySensor(BinarySensorEntity):
if condition.state(self.hass, entity, [STATE_UNKNOWN, STATE_UNAVAILABLE]):
return None
return condition.state(self.hass, entity, entity_observation.to_state)
result = condition.state(self.hass, entity, entity_observation.to_state)
if multi and not result:
return None
except ConditionError:
return None
def _process_multi_state(self, entity_observation: Observation) -> bool | None:
"""Return True if state conditions are met, otherwise return None.
Never return False as all other states should have their own probabilities configured.
"""
entity = entity_observation.entity_id
try:
if condition.state(self.hass, entity, entity_observation.to_state):
return True
except ConditionError:
return None
return None
else:
return result
@property
def extra_state_attributes(self) -> dict[str, Any]:

1
homeassistant/components/bayesian/const.py
View File

@ -8,6 +8,7 @@ ATTR_PROBABILITY_THRESHOLD = "probability_threshold"
CONF_OBSERVATIONS = "observations"
CONF_PRIOR = "prior"
CONF_TEMPLATE = "template"
CONF_NUMERIC_STATE = "numeric_state"
CONF_PROBABILITY_THRESHOLD = "probability_threshold"
CONF_P_GIVEN_F = "prob_given_false"
CONF_P_GIVEN_T = "prob_given_true"

1
homeassistant/components/bayesian/helpers.py
View File

@ -33,6 +33,7 @@ class Observation:
below: float | None
value_template: Template | None
observed: bool | None = None
multi: bool = False
id: uuid.UUID = field(default_factory=uuid.uuid4)
def to_dict(self) -> dict[str, str | float | bool | None]:

400
tests/components/bayesian/test_binary_sensor.py
View File

@ -1,6 +1,7 @@
"""The test for the bayesian sensor platform."""
import json
from logging import WARNING
from unittest.mock import patch
import pytest
@ -20,16 +21,14 @@ from homeassistant.const import (
STATE_UNKNOWN,
)
from homeassistant.core import Context, HomeAssistant, callback
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.setup import async_setup_component
from tests.common import get_fixture_path
async def test_load_values_when_added_to_hass(
hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> None:
async def test_load_values_when_added_to_hass(hass: HomeAssistant) -> None:
"""Test that sensor initializes with observations of relevant entities."""
config = {
@ -58,11 +57,6 @@ async def test_load_values_when_added_to_hass(
assert await async_setup_component(hass, "binary_sensor", config)
await hass.async_block_till_done()
assert (
entity_registry.entities["binary_sensor.test_binary"].unique_id
== "bayesian-3b4c9563-5e84-4167-8fe7-8f507e796d72"
)
state = hass.states.get("binary_sensor.test_binary")
assert state.attributes.get("device_class") == "connectivity"
assert state.attributes.get("observations")[0]["prob_given_true"] == 0.8
@ -331,6 +325,75 @@ async def test_sensor_value_template(hass: HomeAssistant) -> None:
assert state.state == "off"
async def test_mixed_states(hass: HomeAssistant) -> None:
"""Test sensor on probability threshold limits."""
config = {
"binary_sensor": {
"name": "should_HVAC",
"platform": "bayesian",
"observations": [
{
"platform": "template",
"value_template": "{{states('sensor.guest_sensor') != 'off'}}",
"prob_given_true": 0.3,
"prob_given_false": 0.15,
},
{
"platform": "state",
"entity_id": "sensor.anyone_home",
"to_state": "on",
"prob_given_true": 0.6,
"prob_given_false": 0.05,
},
{
"platform": "numeric_state",
"entity_id": "sensor.temperature",
"below": 24,
"above": 19,
"prob_given_true": 0.1,
"prob_given_false": 0.6,
},
],
"prior": 0.3,
"probability_threshold": 0.5,
}
}
assert await async_setup_component(hass, "binary_sensor", config)
await hass.async_block_till_done()
hass.states.async_set("sensor.guest_sensor", "UNKNOWN")
hass.states.async_set("sensor.anyone_home", "on")
hass.states.async_set("sensor.temperature", 15)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.should_HVAC")
assert set(state.attributes.get("occurred_observation_entities")) == {
"sensor.anyone_home",
"sensor.temperature",
}
template_obs = {
"platform": "template",
"value_template": "{{states('sensor.guest_sensor') != 'off'}}",
"prob_given_true": 0.3,
"prob_given_false": 0.15,
"observed": True,
}
assert template_obs in state.attributes.get("observations")
assert abs(0.95857988 - state.attributes.get("probability")) < 0.01
# A = binary_sensor.should_HVAC being TRUE, P(A) being the prior
# B = value_template evaluating to TRUE
# Bayes theorum is P(A|B) = P(B|A) * P(A) / ( P(B|A)*P(A) + P(B|~A)*P(~A) ).
# Calculated where P(A) = 0.3, P(B|A) = 0.3 , P(B|notA) = 0.15 = 0.46153846
# Step 2, prior is now 0.46153846, B now refers to sensor.anyone_home=='on'
# P(A) = 0.46153846, P(B|A) = 0.6 , P(B|notA) = 0.05, result = 0.91139240
# Step 3, prior is now 0.91139240, B now refers to sensor.temperature in range [19,24]
# However since the temp is 15 we take the inverse probability for this negative observation
# P(A) = 0.91139240, P(B|A) = (1-0.1) , P(B|notA) = (1-0.6), result = 0.95857988
async def test_threshold(hass: HomeAssistant, issue_registry: ir.IssueRegistry) -> None:
"""Test sensor on probability threshold limits."""
config = {
@ -367,7 +430,7 @@ async def test_threshold(hass: HomeAssistant, issue_registry: ir.IssueRegistry)
async def test_multiple_observations(hass: HomeAssistant) -> None:
"""Test sensor with multiple observations of same entity.
these entries should be labelled as 'multi_state' and negative observations ignored - as the outcome is not known to be binary.
these entries should be labelled as 'state' and negative observations ignored - as the outcome is not known to be binary.
Before the merge of #67631 this practice was a common work-around for bayesian's ignoring of negative observations,
this also preserves that function
"""
@ -436,83 +499,203 @@ async def test_multiple_observations(hass: HomeAssistant) -> None:
# Calculated using bayes theorum where P(A) = 0.2, P(B|A) = 0.2, P(B|notA) = 0.6
assert state.state == "off"
assert state.attributes.get("observations")[0]["platform"] == "multi_state"
assert state.attributes.get("observations")[1]["platform"] == "multi_state"
assert state.attributes.get("observations")[0]["platform"] == "state"
assert state.attributes.get("observations")[1]["platform"] == "state"
async def test_multiple_numeric_observations(hass: HomeAssistant) -> None:
"""Test sensor with multiple numeric observations of same entity."""
async def test_multiple_numeric_observations(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Test sensor on numeric state platform observations with more than one range.
This tests an example where the probability of it being a 'nice day' varies over
a series of temperatures. Since this is a multi-state, all the non-observed ranges
should be ignored and only the range including the observed value should update
the prior. When a value lands on above or below (15 is tested) it is included if it
equals `below`, and ignored if it equals `above`.
"""
config = {
"binary_sensor": {
"platform": "bayesian",
"name": "Test_Binary",
"name": "nice_day",
"observations": [
{
"platform": "numeric_state",
"entity_id": "sensor.test_monitored",
"below": 10,
"above": 0,
"prob_given_true": 0.4,
"prob_given_false": 0.0001,
"entity_id": "sensor.test_temp",
"below": 0,
"prob_given_true": 0.05,
"prob_given_false": 0.2,
},
{
"platform": "numeric_state",
"entity_id": "sensor.test_monitored",
"below": 100,
"above": 30,
"prob_given_true": 0.6,
"prob_given_false": 0.0001,
"entity_id": "sensor.test_temp",
"below": 10,
"above": 0,
"prob_given_true": 0.1,
"prob_given_false": 0.25,
},
{
"platform": "numeric_state",
"entity_id": "sensor.test_temp",
"below": 15,
"above": 10,
"prob_given_true": 0.2,
"prob_given_false": 0.35,
},
{
"platform": "numeric_state",
"entity_id": "sensor.test_temp",
"below": 25,
"above": 15,
"prob_given_true": 0.5,
"prob_given_false": 0.15,
},
{
"platform": "numeric_state",
"entity_id": "sensor.test_temp",
"above": 25,
"prob_given_true": 0.15,
"prob_given_false": 0.05,
},
],
"prior": 0.1,
"prior": 0.3,
}
}
assert await async_setup_component(hass, "binary_sensor", config)
await hass.async_block_till_done()
hass.states.async_set("sensor.test_monitored", STATE_UNKNOWN)
hass.states.async_set("sensor.test_temp", -5)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_binary")
state = hass.states.get("binary_sensor.nice_day")
for attrs in state.attributes.values():
json.dumps(attrs)
assert state.attributes.get("occurred_observation_entities") == []
assert state.attributes.get("occurred_observation_entities") == ["sensor.test_temp"]
assert state.attributes.get("probability") == 0.1
# No observations made so probability should be the prior
assert state.attributes.get("occurred_observation_entities") == ["sensor.test_temp"]
assert abs(state.attributes.get("probability") - 0.09677) < 0.01
# A = binary_sensor.nice_day being TRUE
# B = sensor.test_temp in the range (, 0]
# Bayes theorum is P(A|B) = P(B|A) * P(A) / ( P(B|A)*P(A) + P(B|~A)*P(~A) ).
# Where P(B|A) is prob_given_true and P(B|~A) is prob_given_false
# Calculated using P(A) = 0.3, P(B|A) = 0.05, P(B|~A) = 0.2 -> 0.09677
# Because >1 range is defined for sensor.test_temp we should not infer anything from the
# ranges not observed
assert state.state == "off"
hass.states.async_set("sensor.test_temp", 5)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.nice_day")
assert state.attributes.get("occurred_observation_entities") == ["sensor.test_temp"]
assert abs(state.attributes.get("probability") - 0.14634146) < 0.01
# A = binary_sensor.nice_day being TRUE
# B = sensor.test_temp in the range (0, 10]
# Bayes theorum is P(A|B) = P(B|A) * P(A) / ( P(B|A)*P(A) + P(B|~A)*P(~A) ).
# Where P(B|A) is prob_given_true and P(B|~A) is prob_given_false
# Calculated using P(A) = 0.3, P(B|A) = 0.1, P(B|~A) = 0.25 -> 0.14634146
# Because >1 range is defined for sensor.test_temp we should not infer anything from the
# ranges not observed
assert state.state == "off"
hass.states.async_set("sensor.test_monitored", 20)
hass.states.async_set("sensor.test_temp", 12)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_binary")
assert state.attributes.get("occurred_observation_entities") == [
"sensor.test_monitored"
]
assert round(abs(0.026 - state.attributes.get("probability")), 7) < 0.01
# Step 1 Calculated where P(A) = 0.1, P(~B|A) = 0.6 (negative obs), P(~B|notA) = 0.9999 -> 0.0625
# Step 2 P(A) = 0.0625, P(B|A) = 0.4 (negative obs), P(B|notA) = 0.9999 -> 0.26
state = hass.states.get("binary_sensor.nice_day")
assert abs(state.attributes.get("probability") - 0.19672131) < 0.01
# A = binary_sensor.nice_day being TRUE
# B = sensor.test_temp in the range (10, 15]
# Bayes theorum is P(A|B) = P(B|A) * P(A) / ( P(B|A)*P(A) + P(B|~A)*P(~A) ).
# Where P(B|A) is prob_given_true and P(B|~A) is prob_given_false
# Calculated using P(A) = 0.3, P(B|A) = 0.2, P(B|~A) = 0.35 -> 0.19672131
# Because >1 range is defined for sensor.test_temp we should not infer anything from the
# ranges not observed
assert state.state == "off"
hass.states.async_set("sensor.test_monitored", 35)
hass.states.async_set("sensor.test_temp", 22)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_binary")
assert state.attributes.get("occurred_observation_entities") == [
"sensor.test_monitored"
]
assert abs(1 - state.attributes.get("probability")) < 0.01
# Step 1 Calculated where P(A) = 0.1, P(~B|A) = 0.6 (negative obs), P(~B|notA) = 0.9999 -> 0.0625
# Step 2 P(A) = 0.0625, P(B|A) = 0.6, P(B|notA) = 0.0001 -> 0.9975
state = hass.states.get("binary_sensor.nice_day")
assert abs(state.attributes.get("probability") - 0.58823529) < 0.01
# A = binary_sensor.nice_day being TRUE
# B = sensor.test_temp in the range (15, 25]
# Bayes theorum is P(A|B) = P(B|A) * P(A) / ( P(B|A)*P(A) + P(B|~A)*P(~A) ).
# Where P(B|A) is prob_given_true and P(B|~A) is prob_given_false
# Calculated using P(A) = 0.3, P(B|A) = 0.5, P(B|~A) = 0.15 -> 0.58823529
# Because >1 range is defined for sensor.test_temp we should not infer anything from the
# ranges not observed
assert state.state == "on"
hass.states.async_set("sensor.test_temp", 30)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.nice_day")
assert abs(state.attributes.get("probability") - 0.562500) < 0.01
# A = binary_sensor.nice_day being TRUE
# B = sensor.test_temp in the range (25, ]
# Bayes theorum is P(A|B) = P(B|A) * P(A) / ( P(B|A)*P(A) + P(B|~A)*P(~A) ).
# Where P(B|A) is prob_given_true and P(B|~A) is prob_given_false
# Calculated using P(A) = 0.3, P(B|A) = 0.15, P(B|~A) = 0.05 -> 0.562500
# Because >1 range is defined for sensor.test_temp we should not infer anything from the
# ranges not observed
assert state.state == "on"
# Edge cases
# if on a threshold only one observation should be included and not both
hass.states.async_set("sensor.test_temp", 15)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.nice_day")
assert state.attributes.get("occurred_observation_entities") == ["sensor.test_temp"]
assert abs(state.attributes.get("probability") - 0.19672131) < 0.01
# Where there are multi numeric ranges when on the threshold, use below
# A = binary_sensor.nice_day being TRUE
# B = sensor.test_temp in the range (10, 15]
# Bayes theorum is P(A|B) = P(B|A) * P(A) / ( P(B|A)*P(A) + P(B|~A)*P(~A) ).
# Where P(B|A) is prob_given_true and P(B|~A) is prob_given_false
# Calculated using P(A) = 0.3, P(B|A) = 0.2, P(B|~A) = 0.35 -> 0.19672131
# Because >1 range is defined for sensor.test_temp we should not infer anything from the
# ranges not observed
assert state.state == "off"
assert len(issue_registry.issues) == 0
assert state.attributes.get("observations")[0]["platform"] == "numeric_state"
assert state.attributes.get("observations")[1]["platform"] == "numeric_state"
hass.states.async_set("sensor.test_temp", "badstate")
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.nice_day")
assert state.attributes.get("occurred_observation_entities") == []
assert state.state == "off"
hass.states.async_set("sensor.test_temp", STATE_UNAVAILABLE)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.nice_day")
assert state.attributes.get("occurred_observation_entities") == []
assert state.state == "off"
hass.states.async_set("sensor.test_temp", STATE_UNKNOWN)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.nice_day")
assert state.attributes.get("occurred_observation_entities") == []
assert state.state == "off"
async def test_mirrored_observations(
@ -651,6 +834,127 @@ async def test_missing_prob_given_false(
)
async def test_bad_multi_numeric(
hass: HomeAssistant,
issue_registry: ir.IssueRegistry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test whether missing prob_given_false are detected and appropriate issues are created."""
config = {
"binary_sensor": {
"platform": "bayesian",
"name": "bins_out",
"observations": [
{
"platform": "numeric_state",
"entity_id": "sensor.signal_strength",
"above": 10,
"prob_given_true": 0.01,
"prob_given_false": 0.3,
},
{
"platform": "numeric_state",
"entity_id": "sensor.signal_strength",
"above": 5,
"below": 10,
"prob_given_true": 0.02,
"prob_given_false": 0.5,
},
{
"platform": "numeric_state",
"entity_id": "sensor.signal_strength",
"above": 0,
"below": 6, # overlaps
"prob_given_true": 0.07,
"prob_given_false": 0.1,
},
{
"platform": "numeric_state",
"entity_id": "sensor.signal_strength",
"above": -10,
"below": 0,
"prob_given_true": 0.3,
"prob_given_false": 0.07,
},
{
"platform": "numeric_state",
"entity_id": "sensor.signal_strength",
"below": -10,
"prob_given_true": 0.6,
"prob_given_false": 0.03,
},
],
"prior": 0.2,
}
}
caplog.clear()
caplog.set_level(WARNING)
assert await async_setup_component(hass, "binary_sensor", config)
assert "entities must not overlap" in caplog.text
async def test_inverted_numeric(
hass: HomeAssistant,
issue_registry: ir.IssueRegistry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test whether missing prob_given_false are detected and appropriate logs are created."""
config = {
"binary_sensor": {
"platform": "bayesian",
"name": "goldilocks_zone",
"observations": [
{
"platform": "numeric_state",
"entity_id": "sensor.temp",
"above": 23,
"below": 20,
"prob_given_true": 0.9,
"prob_given_false": 0.2,
},
],
"prior": 0.4,
}
}
assert await async_setup_component(hass, "binary_sensor", config)
assert (
"bayesian numeric state 'above' (23.0) must be less than 'below' (20.0)"
in caplog.text
)
async def test_no_value_numeric(
hass: HomeAssistant,
issue_registry: ir.IssueRegistry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test whether missing prob_given_false are detected and appropriate logs are created."""
config = {
"binary_sensor": {
"platform": "bayesian",
"name": "goldilocks_zone",
"observations": [
{
"platform": "numeric_state",
"entity_id": "sensor.temp",
"prob_given_true": 0.9,
"prob_given_false": 0.2,
},
],
"prior": 0.4,
}
}
assert await async_setup_component(hass, "binary_sensor", config)
assert "at least one of 'above' or 'below' must be specified" in caplog.text
async def test_probability_updates(hass: HomeAssistant) -> None:
"""Test probability update function."""
prob_given_true = [0.3, 0.6, 0.8]