Refactor setup code in command_line (#114661)

* Refactor setup code in command_line

* Fix rebase

* Review comments
This commit is contained in:
G Johansson 2024-04-02 22:47:04 +03:00 committed by GitHub
parent bf9627ad07
commit 448f8a9568
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 75 additions and 104 deletions

View file

@ -6,33 +6,24 @@ import asyncio
from datetime import datetime, timedelta
from typing import cast
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.const import (
CONF_COMMAND,
CONF_DEVICE_CLASS,
CONF_ICON,
CONF_NAME,
CONF_PAYLOAD_OFF,
CONF_PAYLOAD_ON,
CONF_SCAN_INTERVAL,
CONF_UNIQUE_ID,
CONF_VALUE_TEMPLATE,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.template import Template
from homeassistant.helpers.trigger_template_entity import (
CONF_AVAILABILITY,
ManualTriggerEntity,
)
from homeassistant.helpers.trigger_template_entity import ManualTriggerEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util
from .const import CONF_COMMAND_TIMEOUT, LOGGER
from .const import CONF_COMMAND_TIMEOUT, LOGGER, TRIGGER_ENTITY_OPTIONS
from .sensor import CommandSensorData
DEFAULT_NAME = "Binary Command Sensor"
@ -53,31 +44,24 @@ async def async_setup_platform(
discovery_info = cast(DiscoveryInfoType, discovery_info)
binary_sensor_config = discovery_info
name: str = binary_sensor_config.get(CONF_NAME, DEFAULT_NAME)
command: str = binary_sensor_config[CONF_COMMAND]
payload_off: str = binary_sensor_config[CONF_PAYLOAD_OFF]
payload_on: str = binary_sensor_config[CONF_PAYLOAD_ON]
device_class: BinarySensorDeviceClass | None = binary_sensor_config.get(
CONF_DEVICE_CLASS
)
icon: Template | None = binary_sensor_config.get(CONF_ICON)
value_template: Template | None = binary_sensor_config.get(CONF_VALUE_TEMPLATE)
command_timeout: int = binary_sensor_config[CONF_COMMAND_TIMEOUT]
unique_id: str | None = binary_sensor_config.get(CONF_UNIQUE_ID)
scan_interval: timedelta = binary_sensor_config.get(
CONF_SCAN_INTERVAL, SCAN_INTERVAL
)
availability: Template | None = binary_sensor_config.get(CONF_AVAILABILITY)
if value_template is not None:
if value_template := binary_sensor_config.get(CONF_VALUE_TEMPLATE):
value_template.hass = hass
data = CommandSensorData(hass, command, command_timeout)
trigger_entity_config = {
CONF_UNIQUE_ID: unique_id,
CONF_NAME: Template(name, hass),
CONF_DEVICE_CLASS: device_class,
CONF_ICON: icon,
CONF_AVAILABILITY: availability,
CONF_NAME: Template(binary_sensor_config.get(CONF_NAME, DEFAULT_NAME), hass),
**{
k: v for k, v in binary_sensor_config.items() if k in TRIGGER_ENTITY_OPTIONS
},
}
async_add_entities(

View file

@ -2,7 +2,18 @@
import logging
from homeassistant.const import Platform
from homeassistant.components.sensor import CONF_STATE_CLASS
from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_ICON,
CONF_UNIQUE_ID,
CONF_UNIT_OF_MEASUREMENT,
Platform,
)
from homeassistant.helpers.trigger_template_entity import (
CONF_AVAILABILITY,
CONF_PICTURE,
)
LOGGER = logging.getLogger(__package__)
@ -15,3 +26,13 @@ PLATFORMS = [
Platform.SENSOR,
Platform.SWITCH,
]
TRIGGER_ENTITY_OPTIONS = {
CONF_AVAILABILITY,
CONF_DEVICE_CLASS,
CONF_ICON,
CONF_PICTURE,
CONF_STATE_CLASS,
CONF_UNIQUE_ID,
CONF_UNIT_OF_MEASUREMENT,
}

View file

@ -12,24 +12,19 @@ from homeassistant.const import (
CONF_COMMAND_OPEN,
CONF_COMMAND_STATE,
CONF_COMMAND_STOP,
CONF_ICON,
CONF_NAME,
CONF_SCAN_INTERVAL,
CONF_UNIQUE_ID,
CONF_VALUE_TEMPLATE,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.template import Template
from homeassistant.helpers.trigger_template_entity import (
CONF_AVAILABILITY,
ManualTriggerEntity,
)
from homeassistant.helpers.trigger_template_entity import ManualTriggerEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util, slugify
from .const import CONF_COMMAND_TIMEOUT, LOGGER
from .const import CONF_COMMAND_TIMEOUT, LOGGER, TRIGGER_ENTITY_OPTIONS
from .utils import async_call_shell_with_timeout, async_check_output_or_log
SCAN_INTERVAL = timedelta(seconds=15)
@ -45,30 +40,29 @@ async def async_setup_platform(
covers = []
discovery_info = cast(DiscoveryInfoType, discovery_info)
entities: dict[str, Any] = {slugify(discovery_info[CONF_NAME]): discovery_info}
entities: dict[str, dict[str, Any]] = {
slugify(discovery_info[CONF_NAME]): discovery_info
}
for device_name, device_config in entities.items():
value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE)
if value_template is not None:
for device_name, cover_config in entities.items():
if value_template := cover_config.get(CONF_VALUE_TEMPLATE):
value_template.hass = hass
trigger_entity_config = {
CONF_UNIQUE_ID: device_config.get(CONF_UNIQUE_ID),
CONF_NAME: Template(device_config.get(CONF_NAME, device_name), hass),
CONF_ICON: device_config.get(CONF_ICON),
CONF_AVAILABILITY: device_config.get(CONF_AVAILABILITY),
CONF_NAME: Template(cover_config.get(CONF_NAME, device_name), hass),
**{k: v for k, v in cover_config.items() if k in TRIGGER_ENTITY_OPTIONS},
}
covers.append(
CommandCover(
trigger_entity_config,
device_config[CONF_COMMAND_OPEN],
device_config[CONF_COMMAND_CLOSE],
device_config[CONF_COMMAND_STOP],
device_config.get(CONF_COMMAND_STATE),
cover_config[CONF_COMMAND_OPEN],
cover_config[CONF_COMMAND_CLOSE],
cover_config[CONF_COMMAND_STOP],
cover_config.get(CONF_COMMAND_STATE),
value_template,
device_config[CONF_COMMAND_TIMEOUT],
device_config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL),
cover_config[CONF_COMMAND_TIMEOUT],
cover_config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL),
)
)

View file

@ -8,16 +8,12 @@ from datetime import datetime, timedelta
import json
from typing import Any, cast
from homeassistant.components.sensor import CONF_STATE_CLASS, SensorDeviceClass
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.components.sensor.helpers import async_parse_date_datetime
from homeassistant.const import (
CONF_COMMAND,
CONF_DEVICE_CLASS,
CONF_ICON,
CONF_NAME,
CONF_SCAN_INTERVAL,
CONF_UNIQUE_ID,
CONF_UNIT_OF_MEASUREMENT,
CONF_VALUE_TEMPLATE,
)
from homeassistant.core import HomeAssistant
@ -25,31 +21,17 @@ from homeassistant.exceptions import TemplateError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.template import Template
from homeassistant.helpers.trigger_template_entity import (
CONF_AVAILABILITY,
CONF_PICTURE,
ManualTriggerSensorEntity,
)
from homeassistant.helpers.trigger_template_entity import ManualTriggerSensorEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util
from .const import CONF_COMMAND_TIMEOUT, LOGGER
from .const import CONF_COMMAND_TIMEOUT, LOGGER, TRIGGER_ENTITY_OPTIONS
from .utils import async_check_output_or_log
CONF_JSON_ATTRIBUTES = "json_attributes"
DEFAULT_NAME = "Command Sensor"
TRIGGER_ENTITY_OPTIONS = (
CONF_AVAILABILITY,
CONF_DEVICE_CLASS,
CONF_ICON,
CONF_PICTURE,
CONF_UNIQUE_ID,
CONF_STATE_CLASS,
CONF_UNIT_OF_MEASUREMENT,
)
SCAN_INTERVAL = timedelta(seconds=60)
@ -64,21 +46,19 @@ async def async_setup_platform(
discovery_info = cast(DiscoveryInfoType, discovery_info)
sensor_config = discovery_info
name: str = sensor_config[CONF_NAME]
command: str = sensor_config[CONF_COMMAND]
value_template: Template | None = sensor_config.get(CONF_VALUE_TEMPLATE)
command_timeout: int = sensor_config[CONF_COMMAND_TIMEOUT]
if value_template is not None:
value_template.hass = hass
json_attributes: list[str] | None = sensor_config.get(CONF_JSON_ATTRIBUTES)
scan_interval: timedelta = sensor_config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
data = CommandSensorData(hass, command, command_timeout)
trigger_entity_config = {CONF_NAME: Template(name, hass)}
for key in TRIGGER_ENTITY_OPTIONS:
if key not in sensor_config:
continue
trigger_entity_config[key] = sensor_config[key]
if value_template := sensor_config.get(CONF_VALUE_TEMPLATE):
value_template.hass = hass
trigger_entity_config = {
CONF_NAME: Template(sensor_config[CONF_NAME], hass),
**{k: v for k, v in sensor_config.items() if k in TRIGGER_ENTITY_OPTIONS},
}
async_add_entities(
[

View file

@ -11,24 +11,19 @@ from homeassistant.const import (
CONF_COMMAND_OFF,
CONF_COMMAND_ON,
CONF_COMMAND_STATE,
CONF_ICON,
CONF_NAME,
CONF_SCAN_INTERVAL,
CONF_UNIQUE_ID,
CONF_VALUE_TEMPLATE,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.template import Template
from homeassistant.helpers.trigger_template_entity import (
CONF_AVAILABILITY,
ManualTriggerEntity,
)
from homeassistant.helpers.trigger_template_entity import ManualTriggerEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util, slugify
from .const import CONF_COMMAND_TIMEOUT, LOGGER
from .const import CONF_COMMAND_TIMEOUT, LOGGER, TRIGGER_ENTITY_OPTIONS
from .utils import async_call_shell_with_timeout, async_check_output_or_log
SCAN_INTERVAL = timedelta(seconds=30)
@ -42,34 +37,31 @@ async def async_setup_platform(
) -> None:
"""Find and return switches controlled by shell commands."""
discovery_info = cast(DiscoveryInfoType, discovery_info)
entities: dict[str, Any] = {slugify(discovery_info[CONF_NAME]): discovery_info}
switches = []
discovery_info = cast(DiscoveryInfoType, discovery_info)
entities: dict[str, dict[str, Any]] = {
slugify(discovery_info[CONF_NAME]): discovery_info
}
for object_id, device_config in entities.items():
trigger_entity_config = {
CONF_UNIQUE_ID: device_config.get(CONF_UNIQUE_ID),
CONF_NAME: Template(device_config.get(CONF_NAME, object_id), hass),
CONF_ICON: device_config.get(CONF_ICON),
CONF_AVAILABILITY: device_config.get(CONF_AVAILABILITY),
}
value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE)
if value_template is not None:
for object_id, switch_config in entities.items():
if value_template := switch_config.get(CONF_VALUE_TEMPLATE):
value_template.hass = hass
trigger_entity_config = {
CONF_NAME: Template(switch_config.get(CONF_NAME, object_id), hass),
**{k: v for k, v in switch_config.items() if k in TRIGGER_ENTITY_OPTIONS},
}
switches.append(
CommandSwitch(
trigger_entity_config,
object_id,
device_config[CONF_COMMAND_ON],
device_config[CONF_COMMAND_OFF],
device_config.get(CONF_COMMAND_STATE),
switch_config[CONF_COMMAND_ON],
switch_config[CONF_COMMAND_OFF],
switch_config.get(CONF_COMMAND_STATE),
value_template,
device_config[CONF_COMMAND_TIMEOUT],
device_config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL),
switch_config[CONF_COMMAND_TIMEOUT],
switch_config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL),
)
)