Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from regenmaschine import Authenticator, Client
from regenmaschine.exceptions import HTTPError
from requests.exceptions import ConnectTimeout
conf = config[DOMAIN]
ip_address = conf[CONF_IP_ADDRESS]
password = conf[CONF_PASSWORD]
port = conf[CONF_PORT]
ssl = conf[CONF_SSL]
_LOGGER.debug('Setting up RainMachine client')
try:
auth = Authenticator.create_local(
ip_address, password, port=port, https=ssl)
client = Client(auth)
hass.data[DATA_RAINMACHINE] = RainMachine(client)
except (HTTPError, ConnectTimeout, UnboundLocalError) as exc_info:
_LOGGER.error('An error occurred: %s', str(exc_info))
hass.components.persistent_notification.create(
'Error: {0}<br>'
'You will need to restart hass after fixing.'
''.format(exc_info),
title=NOTIFICATION_TITLE,
notification_id=NOTIFICATION_ID)
return False
_LOGGER.debug('Setting up switch platform')
switch_config = conf.get(CONF_SWITCHES, {})
discovery.load_platform(hass, 'switch', DOMAIN, switch_config, config)
_LOGGER.debug('Setup complete')
async def async_setup_entry(hass, config_entry):
"""Set up RainMachine as config entry."""
_verify_domain_control = verify_domain_control(hass, DOMAIN)
websession = aiohttp_client.async_get_clientsession(hass)
client = Client(websession)
try:
await client.load_local(
config_entry.data[CONF_IP_ADDRESS],
config_entry.data[CONF_PASSWORD],
port=config_entry.data[CONF_PORT],
ssl=config_entry.data[CONF_SSL],
)
except RainMachineError as err:
_LOGGER.error("An error occurred: %s", err)
raise ConfigEntryNotReady
else:
# regenmaschine can load multiple controllers at once, but we only grab the one
# we loaded above:
controller = next(iter(client.controllers.values()))