Split the velbus services code in its own file (#131375)
Maikel Punie
GitHub
parent
0d71828def
commit
00ab5db661
|
|
@ -2,30 +2,22 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import suppress
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
|
||||
from velbusaio.controller import Velbus
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_ADDRESS, CONF_PORT, Platform
|
||||
from homeassistant.core import HomeAssistant, ServiceCall
|
||||
from homeassistant.const import CONF_PORT, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
from homeassistant.helpers import config_validation as cv, device_registry as dr
|
||||
from homeassistant.helpers.storage import STORAGE_DIR
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import (
|
||||
CONF_INTERFACE,
|
||||
CONF_MEMO_TEXT,
|
||||
DOMAIN,
|
||||
SERVICE_CLEAR_CACHE,
|
||||
SERVICE_SCAN,
|
||||
SERVICE_SET_MEMO_TEXT,
|
||||
SERVICE_SYNC,
|
||||
)
|
||||
from .const import DOMAIN
|
||||
from .services import setup_services
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -40,6 +32,8 @@ PLATFORMS = [
|
|||
Platform.SWITCH,
|
||||
]
|
||||
|
||||
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
||||
|
||||
|
||||
async def velbus_connect_task(
|
||||
controller: Velbus, hass: HomeAssistant, entry_id: str
|
||||
|
|
@ -67,6 +61,12 @@ def _migrate_device_identifiers(hass: HomeAssistant, entry_id: str) -> None:
|
|||
dev_reg.async_update_device(device.id, new_identifiers=new_identifier)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the actions for the Velbus component."""
|
||||
setup_services(hass)
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Establish connection with velbus."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
|
|
@ -85,97 +85,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
if hass.services.has_service(DOMAIN, SERVICE_SCAN):
|
||||
return True
|
||||
|
||||
def check_entry_id(interface: str) -> str:
|
||||
for config_entry in hass.config_entries.async_entries(DOMAIN):
|
||||
if "port" in config_entry.data and config_entry.data["port"] == interface:
|
||||
return config_entry.entry_id
|
||||
raise vol.Invalid(
|
||||
"The interface provided is not defined as a port in a Velbus integration"
|
||||
)
|
||||
|
||||
async def scan(call: ServiceCall) -> None:
|
||||
await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].scan()
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_SCAN,
|
||||
scan,
|
||||
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
|
||||
)
|
||||
|
||||
async def syn_clock(call: ServiceCall) -> None:
|
||||
await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].sync_clock()
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_SYNC,
|
||||
syn_clock,
|
||||
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
|
||||
)
|
||||
|
||||
async def set_memo_text(call: ServiceCall) -> None:
|
||||
"""Handle Memo Text service call."""
|
||||
memo_text = call.data[CONF_MEMO_TEXT]
|
||||
await (
|
||||
hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"]
|
||||
.get_module(call.data[CONF_ADDRESS])
|
||||
.set_memo_text(memo_text)
|
||||
)
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_SET_MEMO_TEXT,
|
||||
set_memo_text,
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
|
||||
vol.Required(CONF_ADDRESS): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=0, max=255)
|
||||
),
|
||||
vol.Optional(CONF_MEMO_TEXT, default=""): cv.string,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
async def clear_cache(call: ServiceCall) -> None:
|
||||
"""Handle a clear cache service call."""
|
||||
# clear the cache
|
||||
with suppress(FileNotFoundError):
|
||||
if call.data.get(CONF_ADDRESS):
|
||||
await hass.async_add_executor_job(
|
||||
os.unlink,
|
||||
hass.config.path(
|
||||
STORAGE_DIR,
|
||||
f"velbuscache-{call.data[CONF_INTERFACE]}/{call.data[CONF_ADDRESS]}.p",
|
||||
),
|
||||
)
|
||||
else:
|
||||
await hass.async_add_executor_job(
|
||||
shutil.rmtree,
|
||||
hass.config.path(
|
||||
STORAGE_DIR, f"velbuscache-{call.data[CONF_INTERFACE]}/"
|
||||
),
|
||||
)
|
||||
# call a scan to repopulate
|
||||
await scan(call)
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_CLEAR_CACHE,
|
||||
clear_cache,
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
|
||||
vol.Optional(CONF_ADDRESS): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=0, max=255)
|
||||
),
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
|
|
@ -186,10 +95,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
if not hass.data[DOMAIN]:
|
||||
hass.data.pop(DOMAIN)
|
||||
hass.services.async_remove(DOMAIN, SERVICE_SCAN)
|
||||
hass.services.async_remove(DOMAIN, SERVICE_SYNC)
|
||||
hass.services.async_remove(DOMAIN, SERVICE_SET_MEMO_TEXT)
|
||||
hass.services.async_remove(DOMAIN, SERVICE_CLEAR_CACHE)
|
||||
return unload_ok
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
rules:
|
||||
# Bronze
|
||||
action-setup: todo
|
||||
action-setup: done
|
||||
appropriate-polling:
|
||||
status: exempt
|
||||
comment: |
|
||||
|
|
|
|||
|
|
@ -0,0 +1,116 @@
|
|||
"""Support for Velbus devices."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import suppress
|
||||
import os
|
||||
import shutil
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import CONF_ADDRESS
|
||||
from homeassistant.core import HomeAssistant, ServiceCall
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.storage import STORAGE_DIR
|
||||
|
||||
from .const import (
|
||||
CONF_INTERFACE,
|
||||
CONF_MEMO_TEXT,
|
||||
DOMAIN,
|
||||
SERVICE_CLEAR_CACHE,
|
||||
SERVICE_SCAN,
|
||||
SERVICE_SET_MEMO_TEXT,
|
||||
SERVICE_SYNC,
|
||||
)
|
||||
|
||||
|
||||
def setup_services(hass: HomeAssistant) -> None:
|
||||
"""Register the velbus services."""
|
||||
|
||||
def check_entry_id(interface: str) -> str:
|
||||
for config_entry in hass.config_entries.async_entries(DOMAIN):
|
||||
if "port" in config_entry.data and config_entry.data["port"] == interface:
|
||||
return config_entry.entry_id
|
||||
raise vol.Invalid(
|
||||
"The interface provided is not defined as a port in a Velbus integration"
|
||||
)
|
||||
|
||||
async def scan(call: ServiceCall) -> None:
|
||||
await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].scan()
|
||||
|
||||
async def syn_clock(call: ServiceCall) -> None:
|
||||
await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].sync_clock()
|
||||
|
||||
async def set_memo_text(call: ServiceCall) -> None:
|
||||
"""Handle Memo Text service call."""
|
||||
memo_text = call.data[CONF_MEMO_TEXT]
|
||||
await (
|
||||
hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"]
|
||||
.get_module(call.data[CONF_ADDRESS])
|
||||
.set_memo_text(memo_text.async_render())
|
||||
)
|
||||
|
||||
async def clear_cache(call: ServiceCall) -> None:
|
||||
"""Handle a clear cache service call."""
|
||||
# clear the cache
|
||||
with suppress(FileNotFoundError):
|
||||
if call.data.get(CONF_ADDRESS):
|
||||
await hass.async_add_executor_job(
|
||||
os.unlink,
|
||||
hass.config.path(
|
||||
STORAGE_DIR,
|
||||
f"velbuscache-{call.data[CONF_INTERFACE]}/{call.data[CONF_ADDRESS]}.p",
|
||||
),
|
||||
)
|
||||
else:
|
||||
await hass.async_add_executor_job(
|
||||
shutil.rmtree,
|
||||
hass.config.path(
|
||||
STORAGE_DIR, f"velbuscache-{call.data[CONF_INTERFACE]}/"
|
||||
),
|
||||
)
|
||||
# call a scan to repopulate
|
||||
await scan(call)
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_SCAN,
|
||||
scan,
|
||||
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
|
||||
)
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_SYNC,
|
||||
syn_clock,
|
||||
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
|
||||
)
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_SET_MEMO_TEXT,
|
||||
set_memo_text,
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
|
||||
vol.Required(CONF_ADDRESS): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=0, max=255)
|
||||
),
|
||||
vol.Optional(CONF_MEMO_TEXT, default=""): cv.template,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
SERVICE_CLEAR_CACHE,
|
||||
clear_cache,
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
|
||||
vol.Optional(CONF_ADDRESS): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=0, max=255)
|
||||
),
|
||||
}
|
||||
),
|
||||
)
|
||||
Loading…
Reference in New Issue