164 lines
4.6 KiB
Python
164 lines
4.6 KiB
Python
"""Support for Linky."""
|
|
import json
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from pylinky.client import DAILY, MONTHLY, YEARLY, LinkyClient
|
|
from pylinky.client import PyLinkyException
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import (
|
|
ATTR_ATTRIBUTION,
|
|
CONF_PASSWORD,
|
|
CONF_TIMEOUT,
|
|
CONF_USERNAME,
|
|
ENERGY_KILO_WATT_HOUR,
|
|
)
|
|
from homeassistant.helpers.entity import Entity
|
|
from homeassistant.helpers.event import async_track_time_interval
|
|
from homeassistant.helpers.typing import HomeAssistantType
|
|
|
|
from .const import DOMAIN
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
SCAN_INTERVAL = timedelta(hours=4)
|
|
ICON_ENERGY = "mdi:flash"
|
|
CONSUMPTION = "conso"
|
|
TIME = "time"
|
|
INDEX_CURRENT = -1
|
|
INDEX_LAST = -2
|
|
ATTRIBUTION = "Data provided by Enedis"
|
|
|
|
|
|
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
|
"""Old way of setting up the Linky platform."""
|
|
pass
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
|
|
) -> None:
|
|
"""Add Linky entries."""
|
|
account = LinkyAccount(
|
|
entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD], entry.data[CONF_TIMEOUT]
|
|
)
|
|
|
|
await hass.async_add_executor_job(account.update_linky_data)
|
|
|
|
sensors = [
|
|
LinkySensor("Linky yesterday", account, DAILY, INDEX_LAST),
|
|
LinkySensor("Linky current month", account, MONTHLY, INDEX_CURRENT),
|
|
LinkySensor("Linky last month", account, MONTHLY, INDEX_LAST),
|
|
LinkySensor("Linky current year", account, YEARLY, INDEX_CURRENT),
|
|
LinkySensor("Linky last year", account, YEARLY, INDEX_LAST),
|
|
]
|
|
|
|
async_track_time_interval(hass, account.update_linky_data, SCAN_INTERVAL)
|
|
|
|
async_add_entities(sensors, True)
|
|
|
|
|
|
class LinkyAccount:
|
|
"""Representation of a Linky account."""
|
|
|
|
def __init__(self, username, password, timeout):
|
|
"""Initialise the Linky account."""
|
|
self._username = username
|
|
self._password = password
|
|
self._timeout = timeout
|
|
self._data = None
|
|
|
|
def update_linky_data(self, event_time=None):
|
|
"""Fetch new state data for the sensor."""
|
|
client = LinkyClient(self._username, self._password, None, self._timeout)
|
|
try:
|
|
client.login()
|
|
client.fetch_data()
|
|
self._data = client.get_data()
|
|
_LOGGER.debug(json.dumps(self._data, indent=2))
|
|
except PyLinkyException as exp:
|
|
_LOGGER.error(exp)
|
|
finally:
|
|
client.close_session()
|
|
|
|
@property
|
|
def username(self):
|
|
"""Return the username."""
|
|
return self._username
|
|
|
|
@property
|
|
def data(self):
|
|
"""Return the data."""
|
|
return self._data
|
|
|
|
|
|
class LinkySensor(Entity):
|
|
"""Representation of a sensor entity for Linky."""
|
|
|
|
def __init__(self, name, account: LinkyAccount, scale, when):
|
|
"""Initialize the sensor."""
|
|
self._name = name
|
|
self._account = account
|
|
self._scale = scale
|
|
self._when = when
|
|
self._username = account.username
|
|
self._time = None
|
|
self._consumption = None
|
|
self._unique_id = f"{self._username}_{scale}_{when}"
|
|
|
|
@property
|
|
def unique_id(self):
|
|
"""Return a unique ID."""
|
|
return self._unique_id
|
|
|
|
@property
|
|
def name(self):
|
|
"""Return the name of the sensor."""
|
|
return self._name
|
|
|
|
@property
|
|
def state(self):
|
|
"""Return the state of the sensor."""
|
|
return self._consumption
|
|
|
|
@property
|
|
def unit_of_measurement(self):
|
|
"""Return the unit of measurement."""
|
|
return ENERGY_KILO_WATT_HOUR
|
|
|
|
@property
|
|
def icon(self):
|
|
"""Return the icon of the sensor."""
|
|
return ICON_ENERGY
|
|
|
|
@property
|
|
def device_state_attributes(self):
|
|
"""Return the state attributes of the sensor."""
|
|
return {
|
|
ATTR_ATTRIBUTION: ATTRIBUTION,
|
|
"time": self._time,
|
|
CONF_USERNAME: self._username,
|
|
}
|
|
|
|
@property
|
|
def device_info(self):
|
|
"""Return device information."""
|
|
return {
|
|
"identifiers": {(DOMAIN, self._username)},
|
|
"name": "Linky meter",
|
|
"manufacturer": "Enedis",
|
|
}
|
|
|
|
async def async_update(self) -> None:
|
|
"""Retrieve the new data for the sensor."""
|
|
data = self._account.data[self._scale][self._when]
|
|
self._consumption = data[CONSUMPTION]
|
|
self._time = data[TIME]
|
|
|
|
if self._scale is not YEARLY:
|
|
year_index = INDEX_CURRENT
|
|
if self._time.endswith("Dec"):
|
|
year_index = INDEX_LAST
|
|
self._time += " " + self._account.data[YEARLY][year_index][TIME]
|