Add Netatmo device trigger (#45387)
Co-authored-by: J. Nick Koston <nick@koston.org>pull/47008/head
Tobias Sauerwein
GitHub
parent
db0d815f9d
commit
722b1e8746
|
|
@ -26,12 +26,14 @@ from homeassistant.const import (
|
|||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers import config_validation as cv, entity_platform
|
||||
from homeassistant.helpers.device_registry import async_get_registry
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
|
||||
from .const import (
|
||||
ATTR_HEATING_POWER_REQUEST,
|
||||
ATTR_SCHEDULE_NAME,
|
||||
ATTR_SELECTED_SCHEDULE,
|
||||
DATA_DEVICE_IDS,
|
||||
DATA_HANDLER,
|
||||
DATA_HOMES,
|
||||
DATA_SCHEDULES,
|
||||
|
|
@ -237,6 +239,10 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
|
|||
)
|
||||
)
|
||||
|
||||
registry = await async_get_registry(self.hass)
|
||||
device = registry.async_get_device({(DOMAIN, self._id)}, set())
|
||||
self.hass.data[DOMAIN][DATA_DEVICE_IDS][self._home_id] = device.id
|
||||
|
||||
async def handle_event(self, event):
|
||||
"""Handle webhook events."""
|
||||
data = event["data"]
|
||||
|
|
|
|||
|
|
@ -4,21 +4,36 @@ API = "api"
|
|||
DOMAIN = "netatmo"
|
||||
MANUFACTURER = "Netatmo"
|
||||
|
||||
MODEL_NAPLUG = "Relay"
|
||||
MODEL_NATHERM1 = "Smart Thermostat"
|
||||
MODEL_NRV = "Smart Radiator Valves"
|
||||
MODEL_NOC = "Smart Outdoor Camera"
|
||||
MODEL_NACAMERA = "Smart Indoor Camera"
|
||||
MODEL_NSD = "Smart Smoke Alarm"
|
||||
MODEL_NACAMDOORTAG = "Smart Door and Window Sensors"
|
||||
MODEL_NHC = "Smart Indoor Air Quality Monitor"
|
||||
MODEL_NAMAIN = "Smart Home Weather station – indoor module"
|
||||
MODEL_NAMODULE1 = "Smart Home Weather station – outdoor module"
|
||||
MODEL_NAMODULE4 = "Smart Additional Indoor module"
|
||||
MODEL_NAMODULE3 = "Smart Rain Gauge"
|
||||
MODEL_NAMODULE2 = "Smart Anemometer"
|
||||
MODEL_PUBLIC = "Public Weather stations"
|
||||
|
||||
MODELS = {
|
||||
"NAPlug": "Relay",
|
||||
"NATherm1": "Smart Thermostat",
|
||||
"NRV": "Smart Radiator Valves",
|
||||
"NACamera": "Smart Indoor Camera",
|
||||
"NOC": "Smart Outdoor Camera",
|
||||
"NSD": "Smart Smoke Alarm",
|
||||
"NACamDoorTag": "Smart Door and Window Sensors",
|
||||
"NHC": "Smart Indoor Air Quality Monitor",
|
||||
"NAMain": "Smart Home Weather station – indoor module",
|
||||
"NAModule1": "Smart Home Weather station – outdoor module",
|
||||
"NAModule4": "Smart Additional Indoor module",
|
||||
"NAModule3": "Smart Rain Gauge",
|
||||
"NAModule2": "Smart Anemometer",
|
||||
"public": "Public Weather stations",
|
||||
"NAPlug": MODEL_NAPLUG,
|
||||
"NATherm1": MODEL_NATHERM1,
|
||||
"NRV": MODEL_NRV,
|
||||
"NACamera": MODEL_NACAMERA,
|
||||
"NOC": MODEL_NOC,
|
||||
"NSD": MODEL_NSD,
|
||||
"NACamDoorTag": MODEL_NACAMDOORTAG,
|
||||
"NHC": MODEL_NHC,
|
||||
"NAMain": MODEL_NAMAIN,
|
||||
"NAModule1": MODEL_NAMODULE1,
|
||||
"NAModule4": MODEL_NAMODULE4,
|
||||
"NAModule3": MODEL_NAMODULE3,
|
||||
"NAModule2": MODEL_NAMODULE2,
|
||||
"public": MODEL_PUBLIC,
|
||||
}
|
||||
|
||||
AUTH = "netatmo_auth"
|
||||
|
|
@ -76,12 +91,66 @@ SERVICE_SET_SCHEDULE = "set_schedule"
|
|||
SERVICE_SET_PERSONS_HOME = "set_persons_home"
|
||||
SERVICE_SET_PERSON_AWAY = "set_person_away"
|
||||
|
||||
# Climate events
|
||||
EVENT_TYPE_SET_POINT = "set_point"
|
||||
EVENT_TYPE_CANCEL_SET_POINT = "cancel_set_point"
|
||||
EVENT_TYPE_THERM_MODE = "therm_mode"
|
||||
# Camera events
|
||||
EVENT_TYPE_LIGHT_MODE = "light_mode"
|
||||
EVENT_TYPE_CAMERA_OUTDOOR = "outdoor"
|
||||
EVENT_TYPE_CAMERA_ANIMAL = "animal"
|
||||
EVENT_TYPE_CAMERA_HUMAN = "human"
|
||||
EVENT_TYPE_CAMERA_VEHICLE = "vehicle"
|
||||
EVENT_TYPE_CAMERA_MOVEMENT = "movement"
|
||||
EVENT_TYPE_CAMERA_PERSON = "person"
|
||||
EVENT_TYPE_CAMERA_PERSON_AWAY = "person_away"
|
||||
# Door tags
|
||||
EVENT_TYPE_DOOR_TAG_SMALL_MOVE = "tag_small_move"
|
||||
EVENT_TYPE_DOOR_TAG_BIG_MOVE = "tag_big_move"
|
||||
EVENT_TYPE_DOOR_TAG_OPEN = "tag_open"
|
||||
EVENT_TYPE_OFF = "off"
|
||||
EVENT_TYPE_ON = "on"
|
||||
EVENT_TYPE_SET_POINT = "set_point"
|
||||
EVENT_TYPE_THERM_MODE = "therm_mode"
|
||||
EVENT_TYPE_ALARM_STARTED = "alarm_started"
|
||||
|
||||
OUTDOOR_CAMERA_TRIGGERS = [
|
||||
EVENT_TYPE_CAMERA_ANIMAL,
|
||||
EVENT_TYPE_CAMERA_HUMAN,
|
||||
EVENT_TYPE_CAMERA_OUTDOOR,
|
||||
EVENT_TYPE_CAMERA_VEHICLE,
|
||||
]
|
||||
INDOOR_CAMERA_TRIGGERS = [
|
||||
EVENT_TYPE_CAMERA_MOVEMENT,
|
||||
EVENT_TYPE_CAMERA_PERSON,
|
||||
EVENT_TYPE_CAMERA_PERSON_AWAY,
|
||||
EVENT_TYPE_ALARM_STARTED,
|
||||
]
|
||||
DOOR_TAG_TRIGGERS = [
|
||||
EVENT_TYPE_DOOR_TAG_SMALL_MOVE,
|
||||
EVENT_TYPE_DOOR_TAG_BIG_MOVE,
|
||||
EVENT_TYPE_DOOR_TAG_OPEN,
|
||||
]
|
||||
CLIMATE_TRIGGERS = [
|
||||
EVENT_TYPE_SET_POINT,
|
||||
EVENT_TYPE_CANCEL_SET_POINT,
|
||||
EVENT_TYPE_THERM_MODE,
|
||||
]
|
||||
EVENT_ID_MAP = {
|
||||
EVENT_TYPE_CAMERA_MOVEMENT: "device_id",
|
||||
EVENT_TYPE_CAMERA_PERSON: "device_id",
|
||||
EVENT_TYPE_CAMERA_PERSON_AWAY: "device_id",
|
||||
EVENT_TYPE_CAMERA_ANIMAL: "device_id",
|
||||
EVENT_TYPE_CAMERA_HUMAN: "device_id",
|
||||
EVENT_TYPE_CAMERA_OUTDOOR: "device_id",
|
||||
EVENT_TYPE_CAMERA_VEHICLE: "device_id",
|
||||
EVENT_TYPE_DOOR_TAG_SMALL_MOVE: "device_id",
|
||||
EVENT_TYPE_DOOR_TAG_BIG_MOVE: "device_id",
|
||||
EVENT_TYPE_DOOR_TAG_OPEN: "device_id",
|
||||
EVENT_TYPE_LIGHT_MODE: "device_id",
|
||||
EVENT_TYPE_ALARM_STARTED: "device_id",
|
||||
EVENT_TYPE_CANCEL_SET_POINT: "room_id",
|
||||
EVENT_TYPE_SET_POINT: "room_id",
|
||||
EVENT_TYPE_THERM_MODE: "home_id",
|
||||
}
|
||||
|
||||
MODE_LIGHT_ON = "on"
|
||||
MODE_LIGHT_OFF = "off"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,155 @@
|
|||
"""Provides device automations for Netatmo."""
|
||||
from typing import List
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.automation import AutomationActionType
|
||||
from homeassistant.components.device_automation import TRIGGER_BASE_SCHEMA
|
||||
from homeassistant.components.device_automation.exceptions import (
|
||||
InvalidDeviceAutomationConfig,
|
||||
)
|
||||
from homeassistant.components.homeassistant.triggers import event as event_trigger
|
||||
from homeassistant.const import (
|
||||
ATTR_DEVICE_ID,
|
||||
CONF_DEVICE_ID,
|
||||
CONF_DOMAIN,
|
||||
CONF_ENTITY_ID,
|
||||
CONF_PLATFORM,
|
||||
CONF_TYPE,
|
||||
)
|
||||
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
|
||||
from homeassistant.helpers import config_validation as cv, entity_registry
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from . import DOMAIN
|
||||
from .climate import STATE_NETATMO_AWAY, STATE_NETATMO_HG, STATE_NETATMO_SCHEDULE
|
||||
from .const import (
|
||||
CLIMATE_TRIGGERS,
|
||||
EVENT_TYPE_THERM_MODE,
|
||||
INDOOR_CAMERA_TRIGGERS,
|
||||
MODEL_NACAMERA,
|
||||
MODEL_NATHERM1,
|
||||
MODEL_NOC,
|
||||
MODEL_NRV,
|
||||
NETATMO_EVENT,
|
||||
OUTDOOR_CAMERA_TRIGGERS,
|
||||
)
|
||||
|
||||
CONF_SUBTYPE = "subtype"
|
||||
|
||||
DEVICES = {
|
||||
MODEL_NACAMERA: INDOOR_CAMERA_TRIGGERS,
|
||||
MODEL_NOC: OUTDOOR_CAMERA_TRIGGERS,
|
||||
MODEL_NATHERM1: CLIMATE_TRIGGERS,
|
||||
MODEL_NRV: CLIMATE_TRIGGERS,
|
||||
}
|
||||
|
||||
SUBTYPES = {
|
||||
EVENT_TYPE_THERM_MODE: [
|
||||
STATE_NETATMO_SCHEDULE,
|
||||
STATE_NETATMO_HG,
|
||||
STATE_NETATMO_AWAY,
|
||||
]
|
||||
}
|
||||
|
||||
TRIGGER_TYPES = OUTDOOR_CAMERA_TRIGGERS + INDOOR_CAMERA_TRIGGERS + CLIMATE_TRIGGERS
|
||||
|
||||
TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_ENTITY_ID): cv.entity_id,
|
||||
vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES),
|
||||
vol.Optional(CONF_SUBTYPE): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def async_validate_trigger_config(hass, config):
|
||||
"""Validate config."""
|
||||
config = TRIGGER_SCHEMA(config)
|
||||
|
||||
device_registry = await hass.helpers.device_registry.async_get_registry()
|
||||
device = device_registry.async_get(config[CONF_DEVICE_ID])
|
||||
|
||||
trigger = config[CONF_TYPE]
|
||||
|
||||
if (
|
||||
not device
|
||||
or device.model not in DEVICES
|
||||
or trigger not in DEVICES[device.model]
|
||||
):
|
||||
raise InvalidDeviceAutomationConfig(f"Unsupported model {device.model}")
|
||||
|
||||
return config
|
||||
|
||||
|
||||
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> List[dict]:
|
||||
"""List device triggers for Netatmo devices."""
|
||||
registry = await entity_registry.async_get_registry(hass)
|
||||
device_registry = await hass.helpers.device_registry.async_get_registry()
|
||||
triggers = []
|
||||
|
||||
for entry in entity_registry.async_entries_for_device(registry, device_id):
|
||||
device = device_registry.async_get(device_id)
|
||||
|
||||
for trigger in DEVICES.get(device.model, []):
|
||||
if trigger in SUBTYPES:
|
||||
for subtype in SUBTYPES[trigger]:
|
||||
triggers.append(
|
||||
{
|
||||
CONF_PLATFORM: "device",
|
||||
CONF_DEVICE_ID: device_id,
|
||||
CONF_DOMAIN: DOMAIN,
|
||||
CONF_ENTITY_ID: entry.entity_id,
|
||||
CONF_TYPE: trigger,
|
||||
CONF_SUBTYPE: subtype,
|
||||
}
|
||||
)
|
||||
else:
|
||||
triggers.append(
|
||||
{
|
||||
CONF_PLATFORM: "device",
|
||||
CONF_DEVICE_ID: device_id,
|
||||
CONF_DOMAIN: DOMAIN,
|
||||
CONF_ENTITY_ID: entry.entity_id,
|
||||
CONF_TYPE: trigger,
|
||||
}
|
||||
)
|
||||
|
||||
return triggers
|
||||
|
||||
|
||||
async def async_attach_trigger(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
action: AutomationActionType,
|
||||
automation_info: dict,
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Attach a trigger."""
|
||||
config = TRIGGER_SCHEMA(config)
|
||||
|
||||
device_registry = await hass.helpers.device_registry.async_get_registry()
|
||||
device = device_registry.async_get(config[CONF_DEVICE_ID])
|
||||
|
||||
if not device:
|
||||
return
|
||||
|
||||
if device.model not in DEVICES:
|
||||
return
|
||||
|
||||
event_config = {
|
||||
event_trigger.CONF_PLATFORM: "event",
|
||||
event_trigger.CONF_EVENT_TYPE: NETATMO_EVENT,
|
||||
event_trigger.CONF_EVENT_DATA: {
|
||||
"type": config[CONF_TYPE],
|
||||
ATTR_DEVICE_ID: config[ATTR_DEVICE_ID],
|
||||
},
|
||||
}
|
||||
if config[CONF_TYPE] in SUBTYPES:
|
||||
event_config[event_trigger.CONF_EVENT_DATA]["data"] = {
|
||||
"mode": config[CONF_SUBTYPE]
|
||||
}
|
||||
|
||||
event_config = event_trigger.TRIGGER_SCHEMA(event_config)
|
||||
return await event_trigger.async_attach_trigger(
|
||||
hass, event_config, action, automation_info, platform_type="device"
|
||||
)
|
||||
|
|
@ -5,7 +5,7 @@ from typing import Dict, List
|
|||
from homeassistant.core import CALLBACK_TYPE, callback
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
from .const import DOMAIN, MANUFACTURER, MODELS, SIGNAL_NAME
|
||||
from .const import DATA_DEVICE_IDS, DOMAIN, MANUFACTURER, MODELS, SIGNAL_NAME
|
||||
from .data_handler import NetatmoDataHandler
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
|
@ -58,6 +58,10 @@ class NetatmoBase(Entity):
|
|||
|
||||
await self.data_handler.unregister_data_class(signal_name, None)
|
||||
|
||||
registry = await self.hass.helpers.device_registry.async_get_registry()
|
||||
device = registry.async_get_device({(DOMAIN, self._id)}, set())
|
||||
self.hass.data[DOMAIN][DATA_DEVICE_IDS][self._id] = device.id
|
||||
|
||||
self.async_update_callback()
|
||||
|
||||
async def async_will_remove_from_hass(self):
|
||||
|
|
|
|||
|
|
@ -39,5 +39,27 @@
|
|||
"title": "Netatmo public weather sensor"
|
||||
}
|
||||
}
|
||||
},
|
||||
"device_automation": {
|
||||
"trigger_subtype": {
|
||||
"away": "away",
|
||||
"schedule": "schedule",
|
||||
"hg": "frost guard"
|
||||
},
|
||||
"trigger_type": {
|
||||
"turned_off": "{entity_name} turned off",
|
||||
"turned_on": "{entity_name} turned on",
|
||||
"human": "{entity_name} detected a human",
|
||||
"movement": "{entity_name} detected movement",
|
||||
"person": "{entity_name} detected a person",
|
||||
"person_away": "{entity_name} detected a person has left",
|
||||
"animal": "{entity_name} detected an animal",
|
||||
"outdoor": "{entity_name} detected an outdoor event",
|
||||
"vehicle": "{entity_name} detected a vehicle",
|
||||
"alarm_started": "{entity_name} detected an alarm",
|
||||
"set_point": "Target temperature {entity_name} set manually",
|
||||
"cancel_set_point": "{entity_name} has resumed its schedule",
|
||||
"therm_mode": "{entity_name} switched to \"{subtype}\""
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -15,6 +15,28 @@
|
|||
}
|
||||
}
|
||||
},
|
||||
"device_automation": {
|
||||
"trigger_subtype": {
|
||||
"away": "away",
|
||||
"hg": "frost guard",
|
||||
"schedule": "schedule"
|
||||
},
|
||||
"trigger_type": {
|
||||
"alarm_started": "{entity_name} detected an alarm",
|
||||
"animal": "{entity_name} detected an animal",
|
||||
"cancel_set_point": "{entity_name} has resumed its schedule",
|
||||
"human": "{entity_name} detected a human",
|
||||
"movement": "{entity_name} detected movement",
|
||||
"outdoor": "{entity_name} detected an outdoor event",
|
||||
"person": "{entity_name} detected a person",
|
||||
"person_away": "{entity_name} detected a person has left",
|
||||
"set_point": "Target temperature {entity_name} set manually",
|
||||
"therm_mode": "{entity_name} switched to \"{subtype}\"",
|
||||
"turned_off": "{entity_name} turned off",
|
||||
"turned_on": "{entity_name} turned on",
|
||||
"vehicle": "{entity_name} detected a vehicle"
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"public_weather": {
|
||||
|
|
|
|||
|
|
@ -1,8 +1,7 @@
|
|||
"""The Netatmo integration."""
|
||||
import logging
|
||||
|
||||
from homeassistant.const import ATTR_ID
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ID
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_send
|
||||
|
||||
from .const import (
|
||||
|
|
@ -11,9 +10,11 @@ from .const import (
|
|||
ATTR_IS_KNOWN,
|
||||
ATTR_NAME,
|
||||
ATTR_PERSONS,
|
||||
DATA_DEVICE_IDS,
|
||||
DATA_PERSONS,
|
||||
DEFAULT_PERSON,
|
||||
DOMAIN,
|
||||
EVENT_ID_MAP,
|
||||
NETATMO_EVENT,
|
||||
)
|
||||
|
||||
|
|
@ -38,17 +39,16 @@ async def handle_webhook(hass, webhook_id, request):
|
|||
event_type = data.get(ATTR_EVENT_TYPE)
|
||||
|
||||
if event_type in EVENT_TYPE_MAP:
|
||||
async_send_event(hass, event_type, data)
|
||||
await async_send_event(hass, event_type, data)
|
||||
|
||||
for event_data in data.get(EVENT_TYPE_MAP[event_type], []):
|
||||
async_evaluate_event(hass, event_data)
|
||||
await async_evaluate_event(hass, event_data)
|
||||
|
||||
else:
|
||||
async_evaluate_event(hass, data)
|
||||
await async_evaluate_event(hass, data)
|
||||
|
||||
|
||||
@callback
|
||||
def async_evaluate_event(hass, event_data):
|
||||
async def async_evaluate_event(hass, event_data):
|
||||
"""Evaluate events from webhook."""
|
||||
event_type = event_data.get(ATTR_EVENT_TYPE)
|
||||
|
||||
|
|
@ -62,21 +62,31 @@ def async_evaluate_event(hass, event_data):
|
|||
person_event_data[ATTR_IS_KNOWN] = person.get(ATTR_IS_KNOWN)
|
||||
person_event_data[ATTR_FACE_URL] = person.get(ATTR_FACE_URL)
|
||||
|
||||
async_send_event(hass, event_type, person_event_data)
|
||||
await async_send_event(hass, event_type, person_event_data)
|
||||
|
||||
else:
|
||||
_LOGGER.debug("%s: %s", event_type, event_data)
|
||||
async_send_event(hass, event_type, event_data)
|
||||
await async_send_event(hass, event_type, event_data)
|
||||
|
||||
|
||||
@callback
|
||||
def async_send_event(hass, event_type, data):
|
||||
async def async_send_event(hass, event_type, data):
|
||||
"""Send events."""
|
||||
hass.bus.async_fire(
|
||||
event_type=NETATMO_EVENT, event_data={"type": event_type, "data": data}
|
||||
)
|
||||
_LOGGER.debug("%s: %s", event_type, data)
|
||||
async_dispatcher_send(
|
||||
hass,
|
||||
f"signal-{DOMAIN}-webhook-{event_type}",
|
||||
{"type": event_type, "data": data},
|
||||
)
|
||||
|
||||
if event_type not in EVENT_ID_MAP:
|
||||
return
|
||||
|
||||
data_device_id = data[EVENT_ID_MAP[event_type]]
|
||||
|
||||
hass.bus.async_fire(
|
||||
event_type=NETATMO_EVENT,
|
||||
event_data={
|
||||
"type": event_type,
|
||||
"data": data,
|
||||
ATTR_DEVICE_ID: hass.data[DOMAIN][DATA_DEVICE_IDS].get(data_device_id),
|
||||
},
|
||||
)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,311 @@
|
|||
"""The tests for Netatmo device triggers."""
|
||||
import pytest
|
||||
|
||||
import homeassistant.components.automation as automation
|
||||
from homeassistant.components.netatmo import DOMAIN as NETATMO_DOMAIN
|
||||
from homeassistant.components.netatmo.const import (
|
||||
CLIMATE_TRIGGERS,
|
||||
INDOOR_CAMERA_TRIGGERS,
|
||||
MODEL_NACAMERA,
|
||||
MODEL_NAPLUG,
|
||||
MODEL_NATHERM1,
|
||||
MODEL_NOC,
|
||||
MODEL_NRV,
|
||||
NETATMO_EVENT,
|
||||
OUTDOOR_CAMERA_TRIGGERS,
|
||||
)
|
||||
from homeassistant.components.netatmo.device_trigger import SUBTYPES
|
||||
from homeassistant.const import ATTR_DEVICE_ID
|
||||
from homeassistant.helpers import device_registry
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import (
|
||||
MockConfigEntry,
|
||||
assert_lists_same,
|
||||
async_capture_events,
|
||||
async_get_device_automations,
|
||||
async_mock_service,
|
||||
mock_device_registry,
|
||||
mock_registry,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def device_reg(hass):
|
||||
"""Return an empty, loaded, registry."""
|
||||
return mock_device_registry(hass)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def entity_reg(hass):
|
||||
"""Return an empty, loaded, registry."""
|
||||
return mock_registry(hass)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def calls(hass):
|
||||
"""Track calls to a mock service."""
|
||||
return async_mock_service(hass, "test", "automation")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform,device_type,event_types",
|
||||
[
|
||||
("camera", MODEL_NOC, OUTDOOR_CAMERA_TRIGGERS),
|
||||
("camera", MODEL_NACAMERA, INDOOR_CAMERA_TRIGGERS),
|
||||
("climate", MODEL_NRV, CLIMATE_TRIGGERS),
|
||||
("climate", MODEL_NATHERM1, CLIMATE_TRIGGERS),
|
||||
],
|
||||
)
|
||||
async def test_get_triggers(
|
||||
hass, device_reg, entity_reg, platform, device_type, event_types
|
||||
):
|
||||
"""Test we get the expected triggers from a netatmo devices."""
|
||||
config_entry = MockConfigEntry(domain=NETATMO_DOMAIN, data={})
|
||||
config_entry.add_to_hass(hass)
|
||||
device_entry = device_reg.async_get_or_create(
|
||||
config_entry_id=config_entry.entry_id,
|
||||
connections={(device_registry.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
|
||||
model=device_type,
|
||||
)
|
||||
entity_reg.async_get_or_create(
|
||||
platform, NETATMO_DOMAIN, "5678", device_id=device_entry.id
|
||||
)
|
||||
expected_triggers = []
|
||||
for event_type in event_types:
|
||||
if event_type in SUBTYPES:
|
||||
for subtype in SUBTYPES[event_type]:
|
||||
expected_triggers.append(
|
||||
{
|
||||
"platform": "device",
|
||||
"domain": NETATMO_DOMAIN,
|
||||
"type": event_type,
|
||||
"subtype": subtype,
|
||||
"device_id": device_entry.id,
|
||||
"entity_id": f"{platform}.{NETATMO_DOMAIN}_5678",
|
||||
}
|
||||
)
|
||||
else:
|
||||
expected_triggers.append(
|
||||
{
|
||||
"platform": "device",
|
||||
"domain": NETATMO_DOMAIN,
|
||||
"type": event_type,
|
||||
"device_id": device_entry.id,
|
||||
"entity_id": f"{platform}.{NETATMO_DOMAIN}_5678",
|
||||
}
|
||||
)
|
||||
triggers = [
|
||||
trigger
|
||||
for trigger in await async_get_device_automations(
|
||||
hass, "trigger", device_entry.id
|
||||
)
|
||||
if trigger["domain"] == NETATMO_DOMAIN
|
||||
]
|
||||
assert_lists_same(triggers, expected_triggers)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform,camera_type,event_type",
|
||||
[("camera", MODEL_NOC, trigger) for trigger in OUTDOOR_CAMERA_TRIGGERS]
|
||||
+ [("camera", MODEL_NACAMERA, trigger) for trigger in INDOOR_CAMERA_TRIGGERS]
|
||||
+ [
|
||||
("climate", MODEL_NRV, trigger)
|
||||
for trigger in CLIMATE_TRIGGERS
|
||||
if trigger not in SUBTYPES
|
||||
]
|
||||
+ [
|
||||
("climate", MODEL_NATHERM1, trigger)
|
||||
for trigger in CLIMATE_TRIGGERS
|
||||
if trigger not in SUBTYPES
|
||||
],
|
||||
)
|
||||
async def test_if_fires_on_event(
|
||||
hass, calls, device_reg, entity_reg, platform, camera_type, event_type
|
||||
):
|
||||
"""Test for event triggers firing."""
|
||||
mac_address = "12:34:56:AB:CD:EF"
|
||||
connection = (device_registry.CONNECTION_NETWORK_MAC, mac_address)
|
||||
config_entry = MockConfigEntry(domain=NETATMO_DOMAIN, data={})
|
||||
config_entry.add_to_hass(hass)
|
||||
device_entry = device_reg.async_get_or_create(
|
||||
config_entry_id=config_entry.entry_id,
|
||||
connections={connection},
|
||||
identifiers={(NETATMO_DOMAIN, mac_address)},
|
||||
model=camera_type,
|
||||
)
|
||||
entity_reg.async_get_or_create(
|
||||
platform, NETATMO_DOMAIN, "5678", device_id=device_entry.id
|
||||
)
|
||||
events = async_capture_events(hass, "netatmo_event")
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: [
|
||||
{
|
||||
"trigger": {
|
||||
"platform": "device",
|
||||
"domain": NETATMO_DOMAIN,
|
||||
"device_id": device_entry.id,
|
||||
"entity_id": f"{platform}.{NETATMO_DOMAIN}_5678",
|
||||
"type": event_type,
|
||||
},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {
|
||||
"some": (
|
||||
"{{trigger.event.data.type}} - {{trigger.platform}} - {{trigger.event.data.device_id}}"
|
||||
)
|
||||
},
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
device = device_reg.async_get_device(set(), {connection})
|
||||
assert device is not None
|
||||
|
||||
# Fake that the entity is turning on.
|
||||
hass.bus.async_fire(
|
||||
event_type=NETATMO_EVENT,
|
||||
event_data={
|
||||
"type": event_type,
|
||||
ATTR_DEVICE_ID: device.id,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert len(events) == 1
|
||||
assert len(calls) == 1
|
||||
assert calls[0].data["some"] == f"{event_type} - device - {device.id}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform,camera_type,event_type,sub_type",
|
||||
[
|
||||
("climate", MODEL_NRV, trigger, subtype)
|
||||
for trigger in SUBTYPES
|
||||
for subtype in SUBTYPES[trigger]
|
||||
]
|
||||
+ [
|
||||
("climate", MODEL_NATHERM1, trigger, subtype)
|
||||
for trigger in SUBTYPES
|
||||
for subtype in SUBTYPES[trigger]
|
||||
],
|
||||
)
|
||||
async def test_if_fires_on_event_with_subtype(
|
||||
hass, calls, device_reg, entity_reg, platform, camera_type, event_type, sub_type
|
||||
):
|
||||
"""Test for event triggers firing."""
|
||||
mac_address = "12:34:56:AB:CD:EF"
|
||||
connection = (device_registry.CONNECTION_NETWORK_MAC, mac_address)
|
||||
config_entry = MockConfigEntry(domain=NETATMO_DOMAIN, data={})
|
||||
config_entry.add_to_hass(hass)
|
||||
device_entry = device_reg.async_get_or_create(
|
||||
config_entry_id=config_entry.entry_id,
|
||||
connections={connection},
|
||||
identifiers={(NETATMO_DOMAIN, mac_address)},
|
||||
model=camera_type,
|
||||
)
|
||||
entity_reg.async_get_or_create(
|
||||
platform, NETATMO_DOMAIN, "5678", device_id=device_entry.id
|
||||
)
|
||||
events = async_capture_events(hass, "netatmo_event")
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: [
|
||||
{
|
||||
"trigger": {
|
||||
"platform": "device",
|
||||
"domain": NETATMO_DOMAIN,
|
||||
"device_id": device_entry.id,
|
||||
"entity_id": f"{platform}.{NETATMO_DOMAIN}_5678",
|
||||
"type": event_type,
|
||||
"subtype": sub_type,
|
||||
},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {
|
||||
"some": (
|
||||
"{{trigger.event.data.type}} - {{trigger.event.data.data.mode}} - "
|
||||
"{{trigger.platform}} - {{trigger.event.data.device_id}}"
|
||||
)
|
||||
},
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
device = device_reg.async_get_device(set(), {connection})
|
||||
assert device is not None
|
||||
|
||||
# Fake that the entity is turning on.
|
||||
hass.bus.async_fire(
|
||||
event_type=NETATMO_EVENT,
|
||||
event_data={
|
||||
"type": event_type,
|
||||
"data": {
|
||||
"mode": sub_type,
|
||||
},
|
||||
ATTR_DEVICE_ID: device.id,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert len(events) == 1
|
||||
assert len(calls) == 1
|
||||
assert calls[0].data["some"] == f"{event_type} - {sub_type} - device - {device.id}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform,device_type,event_type",
|
||||
[("climate", MODEL_NAPLUG, trigger) for trigger in CLIMATE_TRIGGERS],
|
||||
)
|
||||
async def test_if_invalid_device(
|
||||
hass, device_reg, entity_reg, platform, device_type, event_type
|
||||
):
|
||||
"""Test for event triggers firing."""
|
||||
mac_address = "12:34:56:AB:CD:EF"
|
||||
connection = (device_registry.CONNECTION_NETWORK_MAC, mac_address)
|
||||
config_entry = MockConfigEntry(domain=NETATMO_DOMAIN, data={})
|
||||
config_entry.add_to_hass(hass)
|
||||
device_entry = device_reg.async_get_or_create(
|
||||
config_entry_id=config_entry.entry_id,
|
||||
connections={connection},
|
||||
identifiers={(NETATMO_DOMAIN, mac_address)},
|
||||
model=device_type,
|
||||
)
|
||||
entity_reg.async_get_or_create(
|
||||
platform, NETATMO_DOMAIN, "5678", device_id=device_entry.id
|
||||
)
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: [
|
||||
{
|
||||
"trigger": {
|
||||
"platform": "device",
|
||||
"domain": NETATMO_DOMAIN,
|
||||
"device_id": device_entry.id,
|
||||
"entity_id": f"{platform}.{NETATMO_DOMAIN}_5678",
|
||||
"type": event_type,
|
||||
},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {
|
||||
"some": (
|
||||
"{{trigger.event.data.type}} - {{trigger.platform}} - {{trigger.event.data.device_id}}"
|
||||
)
|
||||
},
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
Loading…
Reference in New Issue