This is my basic file structure:
.
├── custom_components
│ ├── custom_component
│ └── __init__.py
│ └── auxiliary
│ ├── __init__.py
│ ├── helper.py
│ └── custom_component.py
This is auxiliary/__init__.py:
from logging import getLogger
from voluptuous import Required, Schema
from homeassistant.components.automation import EVENT_AUTOMATION_RELOADED
from homeassistant.const import CONF_ENTITY_ID, CONF_STATE, EVENT_HOMEASSISTANT_STARTED
from homeassistant.core import HomeAssistant, Event
from homeassistant.helpers.config_validation import string
from .helper import create_automations, create_entities_and_automations, CONFIG_INPUT_BOOLEAN, COMPONENT_INPUT_BOOLEAN, \
CONFIG_INPUT_DATETIME, COMPONENT_INPUT_DATETIME, CONFIG_INPUT_NUMBER, COMPONENT_INPUT_NUMBER, CONFIG_INPUT_TEXT, \
COMPONENT_INPUT_TEXT, CONFIG_TIMER, COMPONENT_TIMER
DOMAIN = 'auxiliary'
SCHEMA_SET_STATE = Schema(
{
Required(CONF_ENTITY_ID): string,
Required(CONF_STATE): string
}
)
_LOGGER = getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: dict):
CONFIG_INPUT_BOOLEAN.update(config.get(COMPONENT_INPUT_BOOLEAN, {}))
CONFIG_INPUT_DATETIME.update(config.get(COMPONENT_INPUT_DATETIME, {}))
CONFIG_INPUT_NUMBER.update(config.get(COMPONENT_INPUT_NUMBER, {}))
CONFIG_INPUT_TEXT.update(config.get(COMPONENT_INPUT_TEXT, {}))
CONFIG_TIMER.update(config.get(COMPONENT_TIMER, {}))
async def handle_home_assistant_started_event(event: Event):
await create_entities_and_automations(hass)
async def handle_automation_reload_event(event: Event):
await create_automations(hass)
hass.bus.async_listen(EVENT_HOMEASSISTANT_STARTED, handle_home_assistant_started_event)
hass.bus.async_listen(EVENT_AUTOMATION_RELOADED, handle_automation_reload_event)
return True
and this is auxiliary/helper.py:
from collections import OrderedDict
from datetime import timedelta
from logging import getLogger
from homeassistant.components.automation import async_setup as setup_automation, \
_async_process_config as add_automation, AutomationConfig
from homeassistant.components.input_boolean import async_setup as setup_input_boolean
from homeassistant.components.input_datetime import async_setup as setup_input_datetime
from homeassistant.components.input_number import async_setup as setup_input_number
from homeassistant.components.input_text import async_setup as setup_input_text
from homeassistant.components.timer import async_setup as setup_timer
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.entity_platform import async_get_platforms
from homeassistant.util import slugify
_LOGGER = getLogger(__name__)
COMPONENT_AUTOMATION = 'automation'
COMPONENT_INPUT_BOOLEAN = 'input_boolean'
COMPONENT_INPUT_DATETIME = 'input_datetime'
COMPONENT_INPUT_NUMBER = 'input_number'
COMPONENT_INPUT_TEXT = 'input_text'
COMPONENT_TIMER = 'timer'
CUSTOM_ENTITY_COMPONENTS = {}
SETUP_FUNCTION = {
COMPONENT_AUTOMATION: setup_automation
}
AUTOMATIONS = []
# all input_* entities in HA (config + custom)
CONFIG_INPUT_BOOLEAN = {}
CONFIG_INPUT_DATETIME = {}
CONFIG_INPUT_NUMBER = {}
CONFIG_INPUT_TEXT = {}
CONFIG_TIMER = {}
# save whether custom inputs were declared, so they can be added to HA
CUSTOM_INPUT_BOOLEAN = False
CUSTOM_INPUT_DATETIME = False
CUSTOM_INPUT_NUMBER = False
CUSTOM_INPUT_TEXT = False
CUSTOM_TIMER = False
def add_host(device_id: str, host: Entity):
HOSTS[device_id] = host
def get_host(device_id: str) -> Entity:
return HOSTS.get(device_id, None)
async def _get_platform(hass: HomeAssistant, domain: str):
platform_list = async_get_platforms(hass, domain)
for platform in platform_list:
if platform.domain == domain:
return platform
if domain not in CUSTOM_ENTITY_COMPONENTS:
await SETUP_FUNCTION[domain](hass, {})
CUSTOM_ENTITY_COMPONENTS[domain] = EntityComponent(
_LOGGER, domain, hass, timedelta(seconds=86400)
)
return CUSTOM_ENTITY_COMPONENTS[domain]
async def create_input_boolean(name: str, icon=None) -> str:
data = {
'name': name
}
if icon:
data['icon'] = icon
internal_name = slugify(name)
CONFIG_INPUT_BOOLEAN[internal_name] = data
global CUSTOM_INPUT_BOOLEAN
CUSTOM_INPUT_BOOLEAN = True
return 'input_boolean.{}'.format(internal_name)
async def create_input_datetime(name: str, has_date: bool, has_time: bool, initial=None,
icon=None) -> str:
data = {
'name': name,
'has_date': has_date,
'has_time': has_time
}
if initial:
data['initial'] = initial
if icon:
data['icon'] = icon
internal_name = slugify(name)
CONFIG_INPUT_DATETIME[internal_name] = OrderedDict(data)
global CUSTOM_INPUT_DATETIME
CUSTOM_INPUT_DATETIME = True
return 'input_datetime.{}'.format(internal_name)
async def create_input_number(name: str, _min: int, _max: int, step: int, mode: str,
unit_of_measurement: str, icon=None) -> str:
data = {
'name': name,
'min': _min,
'max': _max,
'step': step,
'mode': mode,
'unit_of_measurement': unit_of_measurement
}
if icon:
data['icon'] = icon
internal_name = slugify(name)
CONFIG_INPUT_NUMBER[internal_name] = data
global CUSTOM_INPUT_NUMBER
CUSTOM_INPUT_NUMBER = True
return 'input_number.{}'.format(internal_name)
async def create_input_text(name: str, _min=0, _max=100, initial=None,
pattern='', mode='text', icon=None) -> str:
data = {
'name': name,
'min': _min,
'max': _max,
'initial': initial,
'pattern': pattern,
'mode': mode
}
if icon:
data['icon'] = icon
internal_name = slugify(name)
CONFIG_INPUT_TEXT[internal_name] = data
global CUSTOM_INPUT_TEXT
CUSTOM_INPUT_TEXT = True
return 'input_text.{}'.format(internal_name)
async def create_timer(name: str, duration='00:00:00') -> str:
data = {
'name': name,
'duration': duration
}
internal_name = slugify(name)
CONFIG_TIMER[internal_name] = data
global CUSTOM_TIMER
CUSTOM_TIMER = True
return 'timer.{}'.format(internal_name)
async def create_entities_and_automations(hass: HomeAssistant):
if CUSTOM_INPUT_BOOLEAN:
for entity_id in list(
filter(lambda eid: COMPONENT_INPUT_BOOLEAN == eid.split('.')[0], hass.states.async_entity_ids())):
hass.states.async_remove(entity_id)
await setup_input_boolean(hass, {COMPONENT_INPUT_BOOLEAN: CONFIG_INPUT_BOOLEAN})
if CUSTOM_INPUT_DATETIME:
for entity_id in list(
filter(lambda eid: COMPONENT_INPUT_DATETIME == eid.split('.')[0], hass.states.async_entity_ids())):
hass.states.async_remove(entity_id)
await setup_input_datetime(hass, {COMPONENT_INPUT_DATETIME: CONFIG_INPUT_DATETIME})
if CUSTOM_INPUT_NUMBER:
for entity_id in list(
filter(lambda eid: COMPONENT_INPUT_NUMBER == eid.split('.')[0], hass.states.async_entity_ids())):
hass.states.async_remove(entity_id)
await setup_input_number(hass, {COMPONENT_INPUT_NUMBER: CONFIG_INPUT_NUMBER})
if CUSTOM_INPUT_TEXT:
for entity_id in list(
filter(lambda eid: COMPONENT_INPUT_TEXT == eid.split('.')[0], hass.states.async_entity_ids())):
hass.states.async_remove(entity_id)
await setup_input_text(hass, {COMPONENT_INPUT_TEXT: CONFIG_INPUT_TEXT})
if CUSTOM_TIMER:
for entity_id in list(
filter(lambda eid: COMPONENT_TIMER == eid.split('.')[0], hass.states.async_entity_ids())):
hass.states.async_remove(entity_id)
await setup_timer(hass, {COMPONENT_TIMER: CONFIG_TIMER})
await create_automations(hass)
async def create_automation(data: dict):
automation = AutomationConfig(data)
raw_config = dict(data)
automation.raw_config = raw_config
AUTOMATIONS.append(automation)
async def create_automations(hass: HomeAssistant):
platform = await _get_platform(hass, "automation")
data = {
'automation': AUTOMATIONS
}
await add_automation(hass, OrderedDict(data), platform)
and custom_component/__init__.py (in this specific case reminder/__init__.py):
import logging
from datetime import timedelta
from homeassistant.helpers.entity_component import EntityComponent
from .const import DOMAIN, SERVICE_DONE
ENTITY_ID_FORMAT = DOMAIN + ".{}"
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=86400)
async def async_setup(hass, config):
component = hass.data[DOMAIN] = EntityComponent(
_LOGGER, DOMAIN, hass, SCAN_INTERVAL
)
await component.async_setup(config)
component.async_register_entity_service(
SERVICE_DONE, {}, "done",
)
return True
async def async_setup_entry(hass, entry):
return await hass.data[DOMAIN].async_setup_entry(entry)
async def async_unload_entry(hass, entry):
return await hass.data[DOMAIN].async_unload_entry(entry)
and an example for auxiliary/custom_component.py (in this specific case reminder.py). This is a reminder component I wrote which heavily utilizes helper.py:
#!/usr/bin/env python
# -*- coding: utf-8 -*
from collections import OrderedDict
from datetime import datetime
from logging import getLogger
from voluptuous import Optional, Required, Schema
from homeassistant.const import CONF_NAME, CONF_ICON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.config_validation import string, positive_int
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.template import Template
from homeassistant.util import slugify
from .helper import create_input_datetime, create_input_number, create_automation
CONF_FREQUENCY = 'frequency'
CONF_NOTIFY = 'notify'
_LOGGER = getLogger(__name__)
async def async_setup_platform(hass: HomeAssistant, config, async_add_entities, discovery_info=None):
name = config[CONF_NAME]
notify = config.get(CONF_NOTIFY)
icon = config.get(CONF_ICON)
letzte_ausfuehrung_name = name + " Letzte Ausführung"
letzte_ausfuehrung_input_datetime_name = 'input_datetime.' + slugify(letzte_ausfuehrung_name)
erinnern_um_name = name + " Erinnern um"
erinnern_um_input_datetime_name = 'input_datetime.' + slugify(erinnern_um_name)
frequenz_name = name + ' Frequenz'
frequenz_input_number_name = 'input_number.' + slugify(frequenz_name)
pause_name = name + ' Pause'
pause_input_number_name = 'input_number.' + slugify(pause_name)
await create_input_datetime(letzte_ausfuehrung_name, True, False, icon=icon)
await create_input_datetime(erinnern_um_name, False, True, icon=icon)
await create_input_number(frequenz_name, 1, 99999, 1, 'box', 'Tage', icon=icon)
await create_input_number(pause_name, 1, 99999, 1, 'box', 'Tage', icon=icon)
await create_automation_for_notification(name, letzte_ausfuehrung_input_datetime_name,
erinnern_um_input_datetime_name,
frequenz_input_number_name, pause_input_number_name, notify)
await create_automation_for_notification_action(name)
async_add_entities([ReminderEntity(name, notify, letzte_ausfuehrung_input_datetime_name,
erinnern_um_input_datetime_name, frequenz_input_number_name, pause_input_number_name)], False)
async def create_automation_for_notification(name: str, letzte_ausfuehrung_name: str,
erinnern_um_name: str,
frequenz_name: str,
pause_name: str,
notify_list: list):
internal_name = slugify(name)
notification_list = []
for notify in notify_list:
notification_list.append(
{
'service': 'notify.{}'.format(notify),
'data': {
'title': name,
'message': Template(
"{% set tage = (((as_timestamp(now().date()) - state_attr('" + letzte_ausfuehrung_name + "', 'timestamp')) | int /60/60/24) | round(0)) - (states('" + frequenz_name + "') | int) %} " + name + " ausstehend seit {{ 'heute.' if tage == 0 else 'gestern.' if tage == 1 else tage | string + ' Tagen.' }}"),
'data': {
'importance': 'default',
'channel': 'Information',
'tag': internal_name + 'Notification',
'color': 'green',
'actions': [
{
'action': internal_name + 'Done',
'title': 'Erledigt'
}
]
}
}
}
)
data = {
'alias': name,
'trigger': [
{
'platform': 'time',
'at': [erinnern_um_name]
}
],
'condition': [
{
'condition': 'template',
'value_template': Template(
"{{ (((as_timestamp(now().date()) - state_attr('" + letzte_ausfuehrung_name + "', 'timestamp')) | int /60/60/24) | round(0)) >= (states('" + frequenz_name + "') | int) and ((((as_timestamp(now().date()) - state_attr('" + letzte_ausfuehrung_name + "', 'timestamp')) | int /60/60/24) | round(0)) - (states('" + frequenz_name + "') | int)) % (states('" + pause_name + "') | int) == 0 }}")
}
],
'action': notification_list,
'mode': 'single',
'max_exceeded': 'WARNING',
'max': 10,
'trace': {
'stored_traces': 5
}
}
await create_automation(OrderedDict(data))
async def create_automation_for_notification_action(name: str):
internal_name = slugify(name)
data = {
'alias': name + ' erledigt',
'trigger': [
{
'platform': 'event',
'event_type': ['mobile_app_notification_action'],
'event_data': {
'action': internal_name + 'Done'
}
}
],
'action': [
{
'service': 'reminder.done',
'entity_id': 'reminder.{}'.format(internal_name)
}
],
'mode': 'single',
'max_exceeded': 'WARNING',
'max': 10,
'trace': {
'stored_traces': 5
}
}
await create_automation(OrderedDict(data))
class ReminderEntity(Entity):
def __init__(self, name: str, notify: list, letzte_ausfuhrung_input_datetime: str, erinnern_um_input_datetime: str,
frequenz_input_number: str, pause_input_number: str):
self._name = name
self._notify = notify
self._letzte_ausfuhrung_input_datetime = letzte_ausfuhrung_input_datetime
self._erinnern_um_input_datetime = erinnern_um_input_datetime
self._frequenz_input_number = frequenz_input_number
self._pause_input_number = pause_input_number
async def async_update(self):
pass
@property
def name(self):
return self._name
@property
def device_state_attributes(self):
return {
'last_execution': self._letzte_ausfuhrung_input_datetime,
'remind_at': self._erinnern_um_input_datetime,
'frequency': self._frequenz_input_number,
'pause': self._pause_input_number
}
async def done(self):
if self._notify:
data = {
'message': 'clear_notification',
'data': {
'tag': slugify(self._name) + 'Notification'
}
}
for notify in self._notify:
await self.hass.services.async_call('notify', notify, data)
data = {
'entity_id': self._letzte_ausfuhrung_input_datetime,
'date': datetime.now().strftime("%Y-%m-%d")
# 'time': datetime.now().strftime("%H:%M")
}
await self.hass.services.async_call('input_datetime', 'set_datetime', data)
and finally a config entry (reminder.yaml) for this custom component could look like this:
- platform: auxiliary
name: Blumen gießen
notify:
- mobile_app_1
icon: mdi:flower
- platform: auxiliary
name: Haus saugen
notify:
- mobile_app_1
- mobile_app_2
icon: mdi:home-modern
I hope this helps to showcase how I made it work. Sometimes I have to adapt to changes in Home Assistant but this has been stable for a long time now and I built 7 custom components this way.
