add domain folder
This commit is contained in:
vincent committed
1 parent
27596eecd9
commit
0d2aba4b18
15 files changed
+1
-1
No files matched your search
@@ -0,0 +1,206 @@
|
||||
""" 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("Debug", default=False): cv.boolean
|
||||
}
|
||||
)
|
||||
|
||||
if user_input:
|
||||
_LOGGER.debug("config_flow [user] - Step 1b -> On a reçu les valeurs: {}".format(user_input))
|
||||
# Second call; 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"],
|
||||
debug=self._user_inputs["Debug"])
|
||||
try:
|
||||
await self._conn.async_connect()
|
||||
if not self._conn.connected():
|
||||
raise CannotConnectError(reason="Client Modbus not connected")
|
||||
#_LOGGER.debug("test communication with koolnova system")
|
||||
ret, _ = await self._conn.async_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 = {}
|
||||
default_id = 1
|
||||
default_area_name = "Area 1"
|
||||
# set default_id to the last id configured
|
||||
# set default_area_name with the last id configured
|
||||
for area in self._user_inputs["areas"]:
|
||||
default_id = area['Area_id'] + 1
|
||||
default_area_name = "Area " + str(default_id)
|
||||
zone_form = vol.Schema(
|
||||
{
|
||||
vol.Required("Name", default=default_area_name): vol.Coerce(str),
|
||||
vol.Required("Area_id", default=default_id): vol.Coerce(int),
|
||||
vol.Optional("Other_area", default=False): cv.boolean
|
||||
}
|
||||
)
|
||||
|
||||
if user_input:
|
||||
# second call
|
||||
try:
|
||||
# test if area_id is already configured
|
||||
for area in self._user_inputs["areas"]:
|
||||
if user_input['Area_id'] == area['Area_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.async_connect()
|
||||
if user_input['Area_id'] > NB_ZONE_MAX:
|
||||
raise ZoneIdError(reason="Area_id must be between 1 and 16")
|
||||
#_LOGGER.debug("test area registered with id: {}".format(user_input['Area_id']))
|
||||
# test if area is configured into koolnova system
|
||||
ret, _ = await self._conn.async_area_registered(user_input["Area_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['Area_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['Area_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"
|
||||
Reference in new issue
Block a user