Moved to a more simple implementation

This commit is contained in:
Roger Gonzalez 2023-09-23 20:40:25 -03:00
parent d111d187b3
commit 6e40514a2d
Signed by: rogs
GPG Key ID: C7ECE9C6C36EC2E6
4 changed files with 23 additions and 113 deletions

View File

@ -1,84 +0,0 @@
import logging
import re
from typing import Any, Dict, Optional
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_EMAIL
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
CONF_PHONE_NUMBER = "phone_number"
schema = {
vol.Required(CONF_EMAIL): cv.string,
vol.Required(CONF_PHONE_NUMBER): cv.string,
}
AUTH_SCHEMA = vol.Schema(schema)
def validate_email(email: str) -> None:
"""
Validates a email address
Args:
email: The email address to validate.
Raises:
ValueError: If the email address is invalid.
"""
if not re.match(r"[^@]+@[^@]+\.[^@]+", email):
raise ValueError
def validate_uyu_phone_number(phone_number: str) -> None:
"""
Validates a Uruguayan phone number
Args:
phone_number: The phone number to validate.
Raises:
ValueError: If the phone number is invalid.
"""
if not phone_number.startswith("598"):
raise ValueError
if not re.match(r"^[0-9]{11}$", phone_number): # noqa: FS003
raise ValueError
class UTEConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""
UTE Custom config flow.
Args:
config_entries: The config entries.
domain: The domain.
Returns:
The config flow.
"""
data: Optional[Dict[str, Any]]
async def async_step_user(self, user_input: Optional[Dict[str, Any]] = None):
"""Invoked when a user initiates a flow via the user interface."""
errors: Dict[str, str] = {}
if user_input is not None:
try:
validate_email(user_input[CONF_EMAIL])
validate_uyu_phone_number(user_input[CONF_PHONE_NUMBER])
except ValueError:
errors["base"] = "auth"
if not errors:
# Input is valid, set data.
self.data = user_input
return self.async_create_entry(title="UTE", data=self.data)
return self.async_show_form(step_id="user", data_schema=AUTH_SCHEMA, errors=errors)

View File

@ -1 +0,0 @@
DOMAIN = "ute"

View File

@ -4,8 +4,8 @@
"dependencies": [],
"documentation": "https://github.com/rogsme/homeassistant_ute",
"domain": "ute",
"iot_class": "calculated",
"name": "UTE",
"iot_class": "cloud_polling",
"name": "UTE Electricity",
"requirements": ["ute-wrapper==1.0.3"],
"version": "1.0.1"
}

View File

@ -2,36 +2,26 @@ import logging
from datetime import timedelta
from typing import Callable, Optional
from homeassistant import config_entries, core
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import CONF_EMAIL
import homeassistant.helpers.config_validation as cv
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorDeviceClass
from homeassistant.const import CONF_EMAIL, UnitOfPower
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, HomeAssistantType
from ute_wrapper.ute import UTEClient
from .config_flow import CONF_PHONE_NUMBER, schema
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
# Time between updating data from UTE
SCAN_INTERVAL = timedelta(minutes=2)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(schema)
CONF_PHONE_NUMBER = "phone_number"
async def async_setup_entry(
hass: core.HomeAssistant,
config_entry: config_entries.ConfigEntry,
async_add_entities,
):
"""Setup sensors from a config entry created in the integrations UI."""
config = hass.data[DOMAIN][config_entry.entry_id]
ute = UTEClient(
config[CONF_EMAIL],
config[CONF_PHONE_NUMBER],
)
sensor = UTESensor(ute)
async_add_entities(sensor, update_before_add=True)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_EMAIL): cv.string,
vol.Required(CONF_PHONE_NUMBER): cv.string,
}
)
def setup_platform(
@ -41,10 +31,10 @@ def setup_platform(
discovery_info: Optional[DiscoveryInfoType] = None,
) -> None:
"""Set up the sensor platform."""
ute = UTEClient(
config[CONF_EMAIL],
config[CONF_PHONE_NUMBER],
)
email = config[CONF_EMAIL]
phone_number = config[CONF_PHONE_NUMBER]
ute = UTEClient(email, phone_number)
sensor = UTESensor(ute)
async_add_entities(sensor, update_before_add=True)
@ -52,6 +42,11 @@ def setup_platform(
class UTESensor(Entity):
"""Representation of a UTE sensor."""
_attr_name = "UTE Uruguay Client"
_attr_icon = "lightning-bolt"
_attr_native_unit_of_measurement = UnitOfPower.WATT
_attr_device_class = SensorDeviceClass.ENERGY
def __init__(self, ute: UTEClient):
super().__init__()
self.ute = ute
@ -59,6 +54,6 @@ class UTESensor(Entity):
self._available = True
self._name = "Current energy usage"
async def async_update(self):
def update(self):
ute_data = await self.ute.get_current_usage_info()
self._state = ute_data["data"]["power_in_watts"]