199 lines
9.0 KiB
Python
199 lines
9.0 KiB
Python
""" Le Config Flow """
|
|
|
|
import logging
|
|
import voluptuous as vol
|
|
|
|
from homeassistant import exceptions
|
|
import homeassistant.helpers.config_validation as cv
|
|
from homeassistant.config_entries import ConfigFlow
|
|
from homeassistant.data_entry_flow import FlowResult
|
|
from homeassistant.const import CONF_BASE
|
|
from .const import DOMAIN, CONF_NAME
|
|
|
|
from .koolnova.operations import Operations
|
|
from .koolnova.const import (
|
|
DEFAULT_ADDR,
|
|
DEFAULT_BAUDRATE,
|
|
DEFAULT_PARITY,
|
|
DEFAULT_STOPBITS,
|
|
DEFAULT_BYTESIZE,
|
|
NB_ZONE_MAX
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
class TestVBE4ConfigFlow(ConfigFlow, domain=DOMAIN):
|
|
""" La classe qui implémente le config flow notre DOMAIN.
|
|
Elle doit dériver de FlowHandler
|
|
"""
|
|
|
|
# La version de notre configFlow va permettre de migrer les entités
|
|
# vers une version plus récente en cas de changement
|
|
VERSION = 1
|
|
# le dictionnaire qui va recevoir tous les user_input. On le vide au démarrage
|
|
_user_inputs: dict = {}
|
|
_conn = None
|
|
|
|
async def async_step_user(self,
|
|
user_input: dict | None = None) -> FlowResult:
|
|
""" Gestion de l'étape 'user'.
|
|
Point d'entrée de mon configFlow. Cette méthode est appelée 2 fois:
|
|
1. 1ere fois sans user_input -> Affichage du formulaire de configuration
|
|
2. 2eme fois avec les données saisies par l'utilisateur dans user_input -> Sauvegarde des données saisies
|
|
"""
|
|
errors = {}
|
|
user_form = vol.Schema( #pylint: disable=invalid-name
|
|
{
|
|
vol.Required("Name", default="koolnova"): vol.Coerce(str),
|
|
vol.Required("Device", default="/dev/ttyUSB0"): vol.Coerce(str),
|
|
vol.Required("Address", default=DEFAULT_ADDR): vol.Coerce(int),
|
|
vol.Required("Baudrate", default=str(DEFAULT_BAUDRATE)): vol.In(["9600", "19200"]),
|
|
vol.Required("Sizebyte", default=DEFAULT_BYTESIZE): vol.Coerce(int),
|
|
vol.Required("Parity", default="EVEN"): vol.In(["EVEN", "NONE"]),
|
|
vol.Required("Stopbits", default=DEFAULT_STOPBITS): vol.Coerce(int),
|
|
vol.Required("Timeout", default=1): vol.Coerce(int),
|
|
vol.Optional("Discover", default="MANUAL"): vol.In(["MANUAL", "AUTOMATIC"])
|
|
}
|
|
)
|
|
|
|
if user_input:
|
|
# second call
|
|
_LOGGER.debug("config_flow [user] - Step 1b -> On a reçu les valeurs: {}".format(user_input))
|
|
# On memorise les données dans le dictionnaire
|
|
self._user_inputs.update(user_input)
|
|
|
|
self._conn = Operations(port=self._user_inputs["Device"],
|
|
addr=self._user_inputs["Address"],
|
|
baudrate=int(self._user_inputs["Baudrate"]),
|
|
parity=self._user_inputs["Parity"][0],
|
|
bytesize=self._user_inputs["Sizebyte"],
|
|
stopbits=self._user_inputs["Stopbits"],
|
|
timeout=self._user_inputs["Timeout"])
|
|
try:
|
|
await self._conn.connect()
|
|
if not self._conn.connected():
|
|
raise CannotConnectError(reason="Client Modbus not connected")
|
|
_LOGGER.debug("test communication with koolnova system")
|
|
ret, _ = await self._conn.system_status()
|
|
if not ret:
|
|
self._conn.disconnect()
|
|
raise CannotConnectError(reason="Communication error")
|
|
self._conn.disconnect()
|
|
|
|
self._user_inputs["areas"] = []
|
|
# go to next step
|
|
return await self.async_step_areas()
|
|
except CannotConnectError:
|
|
_LOGGER.exception("Cannot connect to koolnova system")
|
|
errors[CONF_BASE] = "cannot_connect"
|
|
except Exception as e:
|
|
_LOGGER.exception("Config Flow generic error")
|
|
|
|
# first call or error
|
|
return self.async_show_form(step_id="user",
|
|
data_schema=user_form,
|
|
errors=errors)
|
|
|
|
async def async_step_areas(self,
|
|
user_input: dict | None = None) -> FlowResult:
|
|
""" Gestion de l'étape de découverte manuelle des zones """
|
|
errors = {}
|
|
zone_form = vol.Schema(
|
|
{
|
|
vol.Required("Name", default="zone"): vol.Coerce(str),
|
|
vol.Required("Zone_id", default=1): vol.Coerce(int),
|
|
vol.Optional("Other_area", default=False): cv.boolean
|
|
}
|
|
)
|
|
|
|
if user_input:
|
|
# second call
|
|
try:
|
|
# test if zone_id is already configured
|
|
for area in self._user_inputs["areas"]:
|
|
if user_input['Zone_id'] == area['Zone_id']:
|
|
raise AreaAlreadySetError(reason="Area is already configured")
|
|
# Last area to configure or not ?
|
|
if not user_input['Other_area']:
|
|
try:
|
|
if not self._conn.connected():
|
|
await self._conn.connect()
|
|
if user_input['Zone_id'] > NB_ZONE_MAX:
|
|
raise ZoneIdError(reason="Zone_Id must be between 1 and 16")
|
|
_LOGGER.debug("test area registered with id: {}".format(user_input['Zone_id']))
|
|
# test if area is configured into koolnova system
|
|
ret, _ = await self._conn.zone_registered(user_input["Zone_id"])
|
|
if not ret:
|
|
self._conn.disconnect()
|
|
raise AreaNotRegistredError(reason="Area Id is not registred")
|
|
|
|
self._conn.disconnect()
|
|
# Update dict
|
|
self._user_inputs["areas"].append(user_input)
|
|
# Create entities
|
|
return self.async_create_entry(title=CONF_NAME,
|
|
data=self._user_inputs)
|
|
except CannotConnectError:
|
|
_LOGGER.exception("Cannot connect to koolnova system")
|
|
errors[CONF_BASE] = "cannot_connect"
|
|
except AreaNotRegistredError:
|
|
_LOGGER.exception("Area (id:{}) is not registered to the koolnova system".format(user_input['Zone_id']))
|
|
errors[CONF_BASE] = "area_not_registered"
|
|
except ZoneIdError:
|
|
_LOGGER.exception("Area Id must be between 1 and 16")
|
|
errors[CONF_BASE] = "zone_id_error"
|
|
except Exception as e:
|
|
_LOGGER.exception("Config Flow generic error")
|
|
else:
|
|
_LOGGER.debug("Config_flow [zone] - Une autre zone à configurer")
|
|
# Update dict
|
|
self._user_inputs["areas"].append(user_input)
|
|
# New area to configure
|
|
return await self.async_step_areas()
|
|
|
|
except AreaAlreadySetError:
|
|
_LOGGER.exception("Area (id:{}) is already configured".format(user_input['Zone_id']))
|
|
errors[CONF_BASE] = "area_already_configured"
|
|
|
|
# first call or error
|
|
return self.async_show_form(step_id="areas",
|
|
data_schema=zone_form,
|
|
errors=errors)
|
|
|
|
class KnownError(exceptions.HomeAssistantError):
|
|
""" Base class for errors known to this config flow
|
|
[error_name] is the value passed to [errors] in async_show_form, which should match
|
|
a key under "errors" in string.json
|
|
|
|
[applies_to_field] is the name of the field name that contains the error (for async_show_form)
|
|
if the field doesn't exist in the form, CONF_BASE will be used instead.
|
|
"""
|
|
error_name = "unknown_error"
|
|
applies_to_field = CONF_BASE
|
|
|
|
def __init__(self, *args: object, **kwargs: dict[str, str]) -> None:
|
|
super().__init__(*args)
|
|
self._extra_info = kwargs
|
|
|
|
def get_errors_and_placeholders(self, schema):
|
|
""" Return dicts of errors and description_placeholders for adding to async_show_form """
|
|
key = self.applies_to_field
|
|
if key not in {k.schema for k in schema}:
|
|
key = CONF_BASE
|
|
return ({key: self.error_name}, self._extra_info or {})
|
|
|
|
class CannotConnectError(KnownError):
|
|
""" Error to indicate we cannot connect """
|
|
error_name = "cannot_connect"
|
|
|
|
class AreaNotRegistredError(KnownError):
|
|
""" Error to indicate that area is not registered """
|
|
error_name = "area_not_registered"
|
|
|
|
class AreaAlreadySetError(KnownError):
|
|
""" Error to indicate that the area is already configured """
|
|
error_name = "area_already_configured"
|
|
|
|
class ZoneIdError(KnownError):
|
|
""" Error with the Zone_Id """
|
|
error_name = "zone_id_error" |