Speed up logbook with a lazy event decoder (#36730)

This commit is contained in:
J. Nick Koston 2020-06-15 12:53:38 -05:00 committed by GitHub
parent 717a21dc7b
commit e443dc1274
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 317 additions and 151 deletions

View File

@ -1,6 +1,7 @@
"""Event parser and human readable log generator."""
from datetime import timedelta
from itertools import groupby
import json
import logging
import time
@ -9,7 +10,7 @@ import voluptuous as vol
from homeassistant.components import sun
from homeassistant.components.http import HomeAssistantView
from homeassistant.components.recorder.models import Events, States
from homeassistant.components.recorder.models import Events, States, process_timestamp
from homeassistant.components.recorder.util import (
QUERY_RETRY_WAIT,
RETRIES,
@ -18,6 +19,7 @@ from homeassistant.components.recorder.util import (
from homeassistant.const import (
ATTR_DOMAIN,
ATTR_ENTITY_ID,
ATTR_FRIENDLY_NAME,
ATTR_HIDDEN,
ATTR_NAME,
CONF_EXCLUDE,
@ -31,7 +33,7 @@ from homeassistant.const import (
STATE_OFF,
STATE_ON,
)
from homeassistant.core import DOMAIN as HA_DOMAIN, State, callback, split_entity_id
from homeassistant.core import DOMAIN as HA_DOMAIN, Context, callback, split_entity_id
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entityfilter import generate_filter
from homeassistant.loader import bind_hass
@ -247,29 +249,34 @@ def humanify(hass, events):
yield data
if event.event_type == EVENT_STATE_CHANGED:
to_state = State.from_dict(event.data.get("new_state"))
new_state = event.data.get("new_state")
domain = to_state.domain
entity_id = new_state.get("entity_id")
domain, object_id = split_entity_id(entity_id)
# Skip all but the last sensor state
if (
domain in CONTINUOUS_DOMAINS
and event != last_sensor_event[to_state.entity_id]
and event != last_sensor_event[entity_id]
):
continue
attributes = new_state.get("attributes", {})
# Don't show continuous sensor value changes in the logbook
if domain in CONTINUOUS_DOMAINS and to_state.attributes.get(
if domain in CONTINUOUS_DOMAINS and attributes.get(
"unit_of_measurement"
):
continue
name = attributes.get(ATTR_FRIENDLY_NAME) or object_id.replace("_", " ")
yield {
"when": event.time_fired,
"name": to_state.name,
"message": _entry_message_from_state(domain, to_state),
"name": name,
"message": _entry_message_from_state(domain, new_state),
"domain": domain,
"entity_id": to_state.entity_id,
"entity_id": entity_id,
"context_id": event.context.id,
"context_user_id": event.context.user_id,
}
@ -303,8 +310,9 @@ def humanify(hass, events):
}
elif event.event_type == EVENT_LOGBOOK_ENTRY:
domain = event.data.get(ATTR_DOMAIN)
entity_id = event.data.get(ATTR_ENTITY_ID)
event_data = event.data
domain = event_data.get(ATTR_DOMAIN)
entity_id = event_data.get(ATTR_ENTITY_ID)
if domain is None and entity_id is not None:
try:
domain = split_entity_id(str(entity_id))[0]
@ -313,8 +321,8 @@ def humanify(hass, events):
yield {
"when": event.time_fired,
"name": event.data.get(ATTR_NAME),
"message": event.data.get(ATTR_MESSAGE),
"name": event_data.get(ATTR_NAME),
"message": event_data.get(ATTR_MESSAGE),
"domain": domain,
"entity_id": entity_id,
"context_id": event.context.id,
@ -375,7 +383,7 @@ def _get_events(hass, config, start_day, end_day, entity_id=None):
def yield_events(query):
"""Yield Events that are not filtered away."""
for row in query.yield_per(500):
event = row.to_native()
event = LazyEvent(row)
if _keep_event(hass, event, entities_filter):
yield event
@ -386,7 +394,13 @@ def _get_events(hass, config, start_day, end_day, entity_id=None):
entity_ids = _get_related_entity_ids(session, entities_filter)
query = (
session.query(Events)
session.query(
Events.event_type,
Events.event_data,
Events.time_fired,
Events.context_id,
Events.context_user_id,
)
.order_by(Events.time_fired)
.outerjoin(States, (Events.event_id == States.event_id))
.filter(
@ -406,8 +420,9 @@ def _get_events(hass, config, start_day, end_day, entity_id=None):
def _keep_event(hass, event, entities_filter):
domain = event.data.get(ATTR_DOMAIN)
entity_id = event.data.get("entity_id")
event_data = event.data
domain = event_data.get(ATTR_DOMAIN)
entity_id = event_data.get("entity_id")
if entity_id:
domain = split_entity_id(entity_id)[0]
@ -416,12 +431,12 @@ def _keep_event(hass, event, entities_filter):
return False
# Do not report on new entities
old_state = event.data.get("old_state")
old_state = event_data.get("old_state")
if old_state is None:
return False
# Do not report on entity removal
new_state = event.data.get("new_state")
new_state = event_data.get("new_state")
if new_state is None:
return False
@ -436,12 +451,11 @@ def _keep_event(hass, event, entities_filter):
return False
# exclude entities which are customized hidden
hidden = attributes.get(ATTR_HIDDEN, False)
if hidden:
if attributes.get(ATTR_HIDDEN, False):
return False
elif event.event_type == EVENT_LOGBOOK_ENTRY:
domain = event.data.get(ATTR_DOMAIN)
domain = event_data.get(ATTR_DOMAIN)
elif not entity_id and event.event_type in hass.data.get(DOMAIN, {}):
# If the entity_id isn't described, use the domain that describes
@ -457,58 +471,61 @@ def _keep_event(hass, event, entities_filter):
def _entry_message_from_state(domain, state):
"""Convert a state to a message for the logbook."""
# We pass domain in so we don't have to split entity_id again
state_state = state.get("state")
if domain in ["device_tracker", "person"]:
if state.state == STATE_NOT_HOME:
if state_state == STATE_NOT_HOME:
return "is away"
return f"is at {state.state}"
return f"is at {state_state}"
if domain == "sun":
if state.state == sun.STATE_ABOVE_HORIZON:
if state_state == sun.STATE_ABOVE_HORIZON:
return "has risen"
return "has set"
device_class = state.attributes.get("device_class")
device_class = state.get("attributes", {}).get("device_class")
if domain == "binary_sensor" and device_class:
if device_class == "battery":
if state.state == STATE_ON:
if state_state == STATE_ON:
return "is low"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "is normal"
if device_class == "connectivity":
if state.state == STATE_ON:
if state_state == STATE_ON:
return "is connected"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "is disconnected"
if device_class in ["door", "garage_door", "opening", "window"]:
if state.state == STATE_ON:
if state_state == STATE_ON:
return "is opened"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "is closed"
if device_class == "lock":
if state.state == STATE_ON:
if state_state == STATE_ON:
return "is unlocked"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "is locked"
if device_class == "plug":
if state.state == STATE_ON:
if state_state == STATE_ON:
return "is plugged in"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "is unplugged"
if device_class == "presence":
if state.state == STATE_ON:
if state_state == STATE_ON:
return "is at home"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "is away"
if device_class == "safety":
if state.state == STATE_ON:
if state_state == STATE_ON:
return "is unsafe"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "is safe"
if device_class in [
@ -525,16 +542,59 @@ def _entry_message_from_state(domain, state):
"sound",
"vibration",
]:
if state.state == STATE_ON:
if state_state == STATE_ON:
return f"detected {device_class}"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return f"cleared (no {device_class} detected)"
if state.state == STATE_ON:
if state_state == STATE_ON:
# Future: combine groups and its entity entries ?
return "turned on"
if state.state == STATE_OFF:
if state_state == STATE_OFF:
return "turned off"
return f"changed to {state.state}"
return f"changed to {state_state}"
class LazyEvent:
"""A lazy version of core Event."""
__slots__ = ["_row", "_event_data", "_time_fired", "_context"]
def __init__(self, row):
"""Init the lazy event."""
self._row = row
self._event_data = None
self._time_fired = None
self._context = None
@property
def event_type(self):
"""Type of event."""
return self._row.event_type
@property
def context(self):
"""Context the event was called."""
if not self._context:
self._context = Context(
id=self._row.context_id, user_id=self._row.context_user_id
)
return self._context
@property
def data(self):
"""Event data."""
if not self._event_data:
self._event_data = json.loads(self._row.event_data)
return self._event_data
@property
def time_fired(self):
"""Time event was fired in utc."""
if not self._time_fired:
self._time_fired = (
process_timestamp(self._row.time_fired) or dt_util.utcnow()
)
return self._time_fired

View File

@ -556,20 +556,26 @@ class TestComponentLogbook(unittest.TestCase):
# message for a device state change
eventA = self.create_state_changed_event(pointA, "switch.bla", 10)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "changed to 10"
# message for a switch turned on
eventA = self.create_state_changed_event(pointA, "switch.bla", STATE_ON)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "turned on"
# message for a switch turned off
eventA = self.create_state_changed_event(pointA, "switch.bla", STATE_OFF)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "turned off"
def test_entry_message_from_state_device_tracker(self):
@ -580,14 +586,18 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "device_tracker.john", STATE_NOT_HOME
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is away"
# message for a device tracker "home" state
eventA = self.create_state_changed_event(pointA, "device_tracker.john", "work")
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is at work"
def test_entry_message_from_state_person(self):
@ -596,14 +606,18 @@ class TestComponentLogbook(unittest.TestCase):
# message for a device tracker "not home" state
eventA = self.create_state_changed_event(pointA, "person.john", STATE_NOT_HOME)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is away"
# message for a device tracker "home" state
eventA = self.create_state_changed_event(pointA, "person.john", "work")
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is at work"
def test_entry_message_from_state_sun(self):
@ -614,16 +628,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "sun.sun", sun.STATE_ABOVE_HORIZON
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "has risen"
# message for a sun set
eventA = self.create_state_changed_event(
pointA, "sun.sun", sun.STATE_BELOW_HORIZON
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "has set"
def test_entry_message_from_state_binary_sensor_battery(self):
@ -635,16 +653,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.battery", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is low"
# message for a binary_sensor battery "normal" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.battery", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is normal"
def test_entry_message_from_state_binary_sensor_connectivity(self):
@ -656,16 +678,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.connectivity", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is connected"
# message for a binary_sensor connectivity "disconnected" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.connectivity", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is disconnected"
def test_entry_message_from_state_binary_sensor_door(self):
@ -677,16 +703,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.door", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is opened"
# message for a binary_sensor door "closed" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.door", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is closed"
def test_entry_message_from_state_binary_sensor_garage_door(self):
@ -698,16 +728,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.garage_door", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is opened"
# message for a binary_sensor garage_door "closed" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.garage_door", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is closed"
def test_entry_message_from_state_binary_sensor_opening(self):
@ -719,16 +753,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.opening", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is opened"
# message for a binary_sensor opening "closed" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.opening", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is closed"
def test_entry_message_from_state_binary_sensor_window(self):
@ -740,16 +778,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.window", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is opened"
# message for a binary_sensor window "closed" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.window", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is closed"
def test_entry_message_from_state_binary_sensor_lock(self):
@ -761,16 +803,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.lock", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is unlocked"
# message for a binary_sensor lock "locked" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.lock", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is locked"
def test_entry_message_from_state_binary_sensor_plug(self):
@ -782,16 +828,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.plug", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is plugged in"
# message for a binary_sensor plug "pluged" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.plug", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is unplugged"
def test_entry_message_from_state_binary_sensor_presence(self):
@ -803,16 +853,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.presence", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is at home"
# message for a binary_sensor presence "away" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.presence", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is away"
def test_entry_message_from_state_binary_sensor_safety(self):
@ -824,16 +878,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.safety", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is unsafe"
# message for a binary_sensor safety "safe" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.safety", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "is safe"
def test_entry_message_from_state_binary_sensor_cold(self):
@ -845,16 +903,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.cold", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected cold"
# message for a binary_sensori cold "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.cold", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no cold detected)"
def test_entry_message_from_state_binary_sensor_gas(self):
@ -866,16 +928,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.gas", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected gas"
# message for a binary_sensori gas "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.gas", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no gas detected)"
def test_entry_message_from_state_binary_sensor_heat(self):
@ -887,16 +953,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.heat", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected heat"
# message for a binary_sensori heat "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.heat", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no heat detected)"
def test_entry_message_from_state_binary_sensor_light(self):
@ -908,16 +978,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.light", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected light"
# message for a binary_sensori light "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.light", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no light detected)"
def test_entry_message_from_state_binary_sensor_moisture(self):
@ -929,16 +1003,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.moisture", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected moisture"
# message for a binary_sensori moisture "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.moisture", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no moisture detected)"
def test_entry_message_from_state_binary_sensor_motion(self):
@ -950,16 +1028,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.motion", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected motion"
# message for a binary_sensori motion "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.motion", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no motion detected)"
def test_entry_message_from_state_binary_sensor_occupancy(self):
@ -971,16 +1053,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.occupancy", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected occupancy"
# message for a binary_sensori occupancy "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.occupancy", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no occupancy detected)"
def test_entry_message_from_state_binary_sensor_power(self):
@ -992,16 +1078,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.power", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected power"
# message for a binary_sensori power "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.power", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no power detected)"
def test_entry_message_from_state_binary_sensor_problem(self):
@ -1013,16 +1103,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.problem", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected problem"
# message for a binary_sensori problem "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.problem", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no problem detected)"
def test_entry_message_from_state_binary_sensor_smoke(self):
@ -1034,16 +1128,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.smoke", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected smoke"
# message for a binary_sensori smoke "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.smoke", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no smoke detected)"
def test_entry_message_from_state_binary_sensor_sound(self):
@ -1055,16 +1153,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.sound", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected sound"
# message for a binary_sensori sound "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.sound", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no sound detected)"
def test_entry_message_from_state_binary_sensor_vibration(self):
@ -1076,16 +1178,20 @@ class TestComponentLogbook(unittest.TestCase):
eventA = self.create_state_changed_event(
pointA, "binary_sensor.vibration", STATE_ON, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "detected vibration"
# message for a binary_sensori vibration "cleared" state
eventA = self.create_state_changed_event(
pointA, "binary_sensor.vibration", STATE_OFF, attributes
)
to_state = ha.State.from_dict(eventA.data.get("new_state"))
message = logbook._entry_message_from_state(to_state.domain, to_state)
new_state = eventA.data.get("new_state")
message = logbook._entry_message_from_state(
ha.split_entity_id(new_state.get("entity_id"))[0], new_state
)
assert message == "cleared (no vibration detected)"
def test_process_custom_logbook_entries(self):