github/ha-emporia-vue
1
1
Fork 0
mirror of https://github.com/magico13/ha-emporia-vue.git synced 2026-01-08 03:53:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #146 from magico13/v8

Browse Source 
V0.8
This commit is contained in:
Mike Marvin
2022-10-08 19:31:17 -04:00
committed by GitHub
parent 469db92399 4574a3e221
commit 1d1199c665
7 changed files with 536 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1,4 +1,5 @@
.vscode
.vs
**/__pycache__
/dist
/build

356
custom_components/emporia_vue/__init__.py
View File

@@ -1,21 +1,30 @@
"""The Emporia Vue integration."""
import asyncio
from datetime import datetime, timedelta, timezone
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from typing import Any
import dateutil.tz
import dateutil.relativedelta
import logging
from pyemvue import PyEmVue
from pyemvue.device import VueDeviceChannel
from pyemvue.device import (
VueDevice,
VueDeviceChannel,
VueUsageDevice,
VueDeviceChannelUsage,
)
from pyemvue.enums import Scale
import re
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, SOURCE_IMPORT
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, VUE_DATA, ENABLE_1M, ENABLE_1D, ENABLE_1MON
@@ -38,11 +47,11 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS = ["sensor", "switch"]
device_gids = []
device_information = {}
last_minute_data = {}
last_day_data = {}
last_day_update = None
DEVICE_GIDS: list[int] = []
DEVICE_INFORMATION: dict[int, VueDevice] = {}
LAST_MINUTE_DATA: dict[str, Any] = {}
LAST_DAY_DATA: dict[str, Any] = {}
LAST_DAY_UPDATE: datetime = None
async def async_setup(hass: HomeAssistant, config: dict):
@@ -70,10 +79,10 @@ async def async_setup(hass: HomeAssistant, config: dict):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up Emporia Vue from a config entry."""
global device_gids
global device_information
device_gids = []
device_information = {}
global DEVICE_GIDS
global DEVICE_INFORMATION
DEVICE_GIDS = []
DEVICE_INFORMATION = {}
entry_data = entry.data
email = entry_data[CONF_EMAIL]
@@ -91,21 +100,22 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
try:
devices = await loop.run_in_executor(None, vue.get_devices)
total_channels = 0
for d in devices:
total_channels += len(d.channels)
_LOGGER.info(
"Found {0} Emporia devices with {1} total channels".format(
len(devices), total_channels
)
)
for device in devices:
if not device.device_gid in device_gids:
device_gids.append(device.device_gid)
if not device.device_gid in DEVICE_GIDS:
DEVICE_GIDS.append(device.device_gid)
# await loop.run_in_executor(None, vue.populate_device_properties, device)
device_information[device.device_gid] = device
DEVICE_INFORMATION[device.device_gid] = device
else:
device_information[device.device_gid].channels += device.channels
DEVICE_INFORMATION[device.device_gid].channels += device.channels
total_channels = 0
for _, device in DEVICE_INFORMATION.items():
total_channels += len(device.channels)
_LOGGER.info(
"Found %s Emporia devices with %s total channels",
len(DEVICE_INFORMATION.keys()),
total_channels,
)
async def async_update_data_1min():
"""Fetch data from API endpoint at a 1 minute interval
@@ -113,12 +123,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
This is the place to pre-process the data to lookup tables
so entities can quickly look up their data.
"""
global last_minute_data
data = await update_sensors(vue, [Scale.MINUTE.value])
# store this, then have the daily sensors pull from it and integrate
# then the daily can "true up" hourly (or more frequent) in case it's incorrect
if data:
last_minute_data = data
global LAST_MINUTE_DATA
LAST_MINUTE_DATA = data
return data
async def async_update_data_1hr():
@@ -130,29 +140,36 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
return await update_sensors(vue, [Scale.MONTH.value])
async def async_update_day_sensors():
global last_day_update
global last_day_data
global LAST_DAY_UPDATE
global LAST_DAY_DATA
now = datetime.now(timezone.utc)
if not last_day_update or (now - last_day_update) > timedelta(minutes=15):
if not LAST_DAY_UPDATE or (now - LAST_DAY_UPDATE) > timedelta(minutes=15):
_LOGGER.info("Updating day sensors")
last_day_update = now
last_day_data = await update_sensors(vue, [Scale.DAY.value])
LAST_DAY_UPDATE = now
LAST_DAY_DATA = await update_sensors(vue, [Scale.DAY.value])
else:
# integrate the minute data
_LOGGER.info("Integrating minute data into day sensors")
if last_minute_data:
for id, data in last_minute_data.items():
day_id = id.rsplit("-", 1)[0] + "-" + Scale.DAY.value
if LAST_MINUTE_DATA:
for identifier, data in LAST_MINUTE_DATA.items():
device_gid, channel_gid, _ = identifier.split("-")
day_id = f"{device_gid}-{channel_gid}-{Scale.DAY.value}"
if (
data
and last_day_data
and last_day_data[day_id]
and last_day_data[day_id]["usage"] is not None
and LAST_DAY_DATA
and day_id in LAST_DAY_DATA
and LAST_DAY_DATA[day_id]
and "usage" in LAST_DAY_DATA[day_id]
and LAST_DAY_DATA[day_id]["usage"] is not None
):
last_day_data[day_id]["usage"] += data[
# if we just passed midnight, then reset back to zero
timestamp: datetime = data["timestamp"]
check_for_midnight(timestamp, int(device_gid), day_id)
LAST_DAY_DATA[day_id]["usage"] += data[
"usage"
] # already in kwh
return last_day_data
return LAST_DAY_DATA
coordinator_1min = None
if ENABLE_1M not in entry_data or entry_data[ENABLE_1M]:
@@ -166,7 +183,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
update_interval=timedelta(minutes=1),
)
await coordinator_1min.async_config_entry_first_refresh()
_LOGGER.info(f"1min Update data: {coordinator_1min.data}")
_LOGGER.info("1min Update data: %s", coordinator_1min.data)
coordinator_1hr = None
if ENABLE_1MON not in entry_data or entry_data[ENABLE_1MON]:
coordinator_1hr = DataUpdateCoordinator(
@@ -179,7 +196,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
update_interval=timedelta(hours=1),
)
await coordinator_1hr.async_config_entry_first_refresh()
_LOGGER.info(f"1hr Update data: {coordinator_1hr.data}")
_LOGGER.info("1hr Update data: %s", coordinator_1hr.data)
coordinator_day_sensor = None
if ENABLE_1D not in entry_data or entry_data[ENABLE_1D]:
@@ -194,8 +211,70 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
)
await coordinator_day_sensor.async_config_entry_first_refresh()
# Setup custom services
async def handle_set_charger_current(call):
"""Handle setting the EV Charger current"""
_LOGGER.debug(
"executing set_charger_current: %s %s",
str(call.service),
str(call.data),
)
current = call.data.get("current")
device_id = call.data.get("device_id", None)
entity_id = call.data.get("entity_id", None)
charger_entity = None
if device_id:
entity_registry = er.async_get(hass)
entities = er.async_entries_for_device(entity_registry, device_id[0])
for entity in entities:
_LOGGER.info("Entity is %s", str(entity))
if entity.entity_id.startswith("switch"):
charger_entity = entity
break
if not charger_entity:
charger_entity = entities[0]
elif entity_id:
entity_registry = er.async_get(hass)
charger_entity = entity_registry.async_get(entity_id[0])
else:
raise HomeAssistantError("Target device or Entity required.")
unique_entity_id = charger_entity.unique_id
gid_match = re.search(r"\d+", unique_entity_id)
if not gid_match:
raise HomeAssistantError(
f"Could not find device gid from unique id {unique_entity_id}"
)
charger_gid = int(gid_match.group(0))
if (
charger_gid not in DEVICE_INFORMATION
or not DEVICE_INFORMATION[charger_gid].ev_charger
):
raise HomeAssistantError(
f"Set Charging Current called on invalid device with entity id {charger_entity.entity_id} (unique id {unique_entity_id})"
)
charger_info = DEVICE_INFORMATION[charger_gid]
# Scale the current to a minimum of 6 amps and max of the circuit max
current = max(6, current)
current = min(current, charger_info.ev_charger.max_charging_rate)
_LOGGER.info(
"Setting charger %s to current of %d amps", charger_gid, current
)
updated_charger = await loop.run_in_executor(
None, vue.update_charger, charger_info.ev_charger, None, current
)
DEVICE_INFORMATION[charger_gid].ev_charger = updated_charger
hass.services.async_register(
DOMAIN, "set_charger_current", handle_set_charger_current
)
except Exception as err:
_LOGGER.warn(f"Exception while setting up Emporia Vue. Will retry. {err}")
_LOGGER.warning("Exception while setting up Emporia Vue. Will retry. %s", err)
raise ConfigEntryNotReady(
f"Exception while setting up Emporia Vue. Will retry. {err}"
)
@@ -213,7 +292,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
hass.config_entries.async_forward_entry_setup(entry, component)
)
except Exception as err:
_LOGGER.warn(f"Error setting up platforms: {err}")
_LOGGER.warning("Error setting up platforms: %s", err)
raise ConfigEntryNotReady(f"Error setting up platforms: {err}")
return True
@@ -235,7 +314,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
return unload_ok
async def update_sensors(vue, scales):
async def update_sensors(vue: PyEmVue, scales: list[str]):
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
@@ -244,53 +323,96 @@ async def update_sensors(vue, scales):
for scale in scales:
utcnow = datetime.now(timezone.utc)
usage_dict = await loop.run_in_executor(
None, vue.get_device_list_usage, device_gids, utcnow, scale
None, vue.get_device_list_usage, DEVICE_GIDS, utcnow, scale
)
if not usage_dict:
_LOGGER.warn(
f"No channels found during update for scale {scale}. Retrying..."
_LOGGER.warning(
"No channels found during update for scale %s. Retrying", scale
)
usage_dict = await loop.run_in_executor(
None, vue.get_device_list_usage, device_gids, utcnow, scale
None, vue.get_device_list_usage, DEVICE_GIDS, utcnow, scale
)
if usage_dict:
recurse_usage_data(usage_dict, scale, data)
recurse_usage_data(usage_dict, scale, data, utcnow)
else:
raise UpdateFailed(f"No channels found during update for scale {scale}")
return data
except Exception as err:
_LOGGER.error(f"Error communicating with Emporia API: {err}")
_LOGGER.error("Error communicating with Emporia API: %s", err)
raise UpdateFailed(f"Error communicating with Emporia API: {err}")
def recurse_usage_data(usage_devices, scale, data):
def recurse_usage_data(
usage_devices: dict[int, VueUsageDevice],
scale: str,
data: dict[str, Any],
requested_time: datetime,
):
"""Loop through the result from get_device_list_usage and pull out the data we want to use."""
for gid, device in usage_devices.items():
for channel_num, channel in device.channels.items():
if not channel:
continue
reset_datetime = None
id = make_channel_id(channel, scale)
info = find_device_info_for_channel(channel)
if scale in [Scale.DAY.value, Scale.MONTH.value]:
reset_datetime = device.timestamp
if device.device_gid in DEVICE_INFORMATION:
info = DEVICE_INFORMATION[device.device_gid]
local_time = change_time_to_local(device.timestamp, info.time_zone)
requested_time_local = change_time_to_local(requested_time, info.time_zone)
if abs((local_time - requested_time_local).total_seconds()) > 30:
_LOGGER.warning(
"More than 30 seconds have passed between the requested datetime and the returned datetime. Requested: %s Returned: %s",
requested_time,
device.timestamp,
)
for channel_num, channel in device.channels.items():
if not channel:
continue
reset_datetime = None
identifier = make_channel_id(channel, scale)
handle_special_channels_for_device(channel)
data[id] = {
"device_gid": gid,
"channel_num": channel_num,
"usage": fix_usage_sign(channel_num, channel.usage),
"scale": scale,
"info": info,
"reset": reset_datetime,
}
if channel.nested_devices:
recurse_usage_data(channel.nested_devices, scale, data)
if scale in [Scale.DAY.value, Scale.MONTH.value]:
# We need to know when the value reset
# For day, that should be midnight local time, but we need to use the timestamp returned to us
# for month, that should be midnight of the reset day they specify in the app
reset_datetime = determine_reset_datetime(
local_time,
info.billing_cycle_start_day,
scale == Scale.MONTH.value,
)
# Fix the usage if we got None
# Use the last value if we have it, otherwise use zero
fixed_usage = channel.usage
if fixed_usage is None:
fixed_usage = handle_none_usage(scale, identifier)
_LOGGER.info(
"Got None usage for device %s channel %s scale %s and timestamp %s. Instead using a value of %s",
gid,
channel_num,
scale,
local_time.isoformat(),
fixed_usage,
)
fixed_usage = fix_usage_sign(channel_num, fixed_usage)
data[identifier] = {
"device_gid": gid,
"channel_num": channel_num,
"usage": fixed_usage,
"scale": scale,
"info": info,
"reset": reset_datetime,
"timestamp": local_time,
}
if channel.nested_devices:
recurse_usage_data(
channel.nested_devices, scale, data, requested_time
)
def find_device_info_for_channel(channel):
def handle_special_channels_for_device(channel: VueDeviceChannelUsage):
device_info = None
if channel.device_gid in device_information:
device_info = device_information[channel.device_gid]
if channel.device_gid in DEVICE_INFORMATION:
device_info = DEVICE_INFORMATION[channel.device_gid]
if channel.channel_num in [
"MainsFromGrid",
"MainsToGrid",
@@ -303,33 +425,99 @@ def find_device_info_for_channel(channel):
if device_channel.channel_num == channel.channel_num:
found = True
break
elif device_channel.channel_num == "1,2,3":
if device_channel.channel_num == "1,2,3":
channel_123 = device_channel
if not found:
_LOGGER.info(
f"Adding channel for channel {channel.device_gid}-{channel.channel_num}"
"Adding channel for channel %s-%s",
channel.device_gid,
channel.channel_num,
)
multiplier = 1.0
type_gid = 1
if channel_123:
multiplier = channel_123.channel_multiplier
type_gid = channel_123.channel_type_gid
device_info.channels.append(
VueDeviceChannel(
gid=channel.device_gid,
name=channel.name,
channelNum=channel.channel_num,
channelMultiplier=channel_123.channel_multiplier,
channelTypeGid=channel_123.channel_type_gid,
channelMultiplier=multiplier,
channelTypeGid=type_gid,
)
)
return device_info
def make_channel_id(channel, scale):
def make_channel_id(channel: VueDeviceChannelUsage, scale: str):
"""Format the channel id for a channel and scale"""
return "{0}-{1}-{2}".format(channel.device_gid, channel.channel_num, scale)
def fix_usage_sign(channel_num, usage):
def fix_usage_sign(channel_num: str, usage: float):
"""If the channel is not '1,2,3' or 'Balance' we need it to be positive (see https://github.com/magico13/ha-emporia-vue/issues/57)"""
if usage and channel_num not in ["1,2,3", "Balance"]:
return abs(usage)
elif not usage:
usage = 0
return usage
def change_time_to_local(time: datetime, tz_string: str):
"""Change the datetime to the provided timezone, if not already."""
tz_info = dateutil.tz.gettz(tz_string)
if not time.tzinfo or time.tzinfo.utcoffset(time) is None:
# unaware, assume it's already utc
time = time.replace(tzinfo=timezone.utc)
return time.astimezone(tz_info)
def check_for_midnight(timestamp: datetime, device_gid: int, day_id: str):
"""If midnight has recently passed, reset the LAST_DAY_DATA for Day sensors to zero"""
if device_gid in DEVICE_INFORMATION:
device_info = DEVICE_INFORMATION[device_gid]
local_time = change_time_to_local(timestamp, device_info.time_zone)
local_midnight = local_time.replace(hour=0, minute=0, second=0, microsecond=0)
last_reset = LAST_DAY_DATA[day_id]["reset"]
if local_midnight > last_reset:
# New reset time found
_LOGGER.warning(
"Midnight happened recently for id %s! Timestamp is %s, midnight is %s, previous reset was %s",
day_id,
local_time,
local_midnight,
last_reset,
)
LAST_DAY_DATA[day_id]["usage"] = 0
LAST_DAY_DATA[day_id]["reset"] = local_midnight
def determine_reset_datetime(
local_time: datetime, monthly_cycle_start: int, is_month: bool
):
"""Determine the last reset datetime (aware) based on the passed time and cycle start date"""
reset_datetime = local_time.replace(hour=0, minute=0, second=0, microsecond=0)
if is_month:
# Month should use the last billing_cycle_start_day of either this or last month
reset_datetime = reset_datetime.replace(day=monthly_cycle_start)
if reset_datetime.day < monthly_cycle_start:
# we're in the start of a month, use the reset_day for last month
reset_datetime -= dateutil.relativedelta.relativedelta(months=1)
return reset_datetime
def handle_none_usage(scale: str, identifier: str):
"""Handle the case of the usage being None by using the previous value or zero."""
if (
scale is Scale.MINUTE.value
and identifier in LAST_MINUTE_DATA
and "usage" in LAST_MINUTE_DATA[identifier]
):
return LAST_MINUTE_DATA[identifier]["usage"]
if (
scale is Scale.DAY.value
and identifier in LAST_DAY_DATA
and "usage" in LAST_DAY_DATA[identifier]
):
return LAST_DAY_DATA[identifier]["usage"]
return 0

74
custom_components/emporia_vue/charger_entity.py Normal file
View File

@@ -0,0 +1,74 @@
import logging
from typing import Any, Mapping
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
)
from .const import DOMAIN
from pyemvue import pyemvue
from pyemvue.device import VueDevice
_LOGGER = logging.getLogger(__name__)
class EmporiaChargerEntity(CoordinatorEntity):
"""Emporia Charger Entity"""
def __init__(
self,
coordinator,
vue: pyemvue.PyEmVue,
device: VueDevice,
units: str,
device_class: str,
enabled_default=True,
):
super().__init__(coordinator)
self._coordinator = coordinator
self._device = device
self._vue = vue
self._enabled_default = enabled_default
self._attr_unit_of_measurement = units
self._attr_device_class = device_class
self._attr_name = device.device_name
@property
def entity_registry_enabled_default(self):
return self._enabled_default
@property
def extra_state_attributes(self) -> Mapping[str, Any]:
data = self._coordinator.data[self._device.device_gid]
if data:
return {
"charging_rate": data.charging_rate,
"max_charging_rate": data.max_charging_rate,
"status": data.status,
"message": data.message,
"fault_text": data.fault_text,
}
return None
@property
def unique_id(self) -> str:
"""Unique ID for the charger"""
return f"charger.emporia_vue.{self._device.device_gid}"
@property
def device_info(self):
"""Return the device information."""
return {
"identifiers": {(DOMAIN, "{0}-1,2,3".format(self._device.device_gid))},
"name": self._device.device_name + "-1,2,3",
"model": self._device.model,
"sw_version": self._device.firmware,
"manufacturer": "Emporia",
}
@property
def available(self):
"""Return True if entity is available."""
return self._device

4
custom_components/emporia_vue/manifest.json
View File

@@ -3,12 +3,12 @@
"name": "Emporia Vue",
"config_flow": true,
"documentation": "https://github.com/magico13/ha-emporia-vue",
"requirements": ["pyemvue==0.15.1"],
"requirements": ["pyemvue==0.16.1"],
"ssdp": [],
"zeroconf": [],
"homekit": {},
"dependencies": [],
"codeowners": ["@magico13"],
"iot_class": "cloud_polling",
"version": "0.7.3"
"version": "0.8.0"
}

106
custom_components/emporia_vue/sensor.py
View File

@@ -1,10 +1,12 @@
"""Platform for sensor integration."""
from homeassistant.components.sensor import STATE_CLASS_TOTAL_INCREASING, STATE_CLASS_MEASUREMENT, SensorEntity
from homeassistant.components.sensor import (
SensorStateClass,
SensorDeviceClass,
SensorEntity,
)
import logging
from homeassistant.const import (
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
POWER_WATT,
ENERGY_KILO_WATT_HOUR,
)
@@ -52,14 +54,14 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class CurrentVuePowerSensor(CoordinatorEntity, SensorEntity):
"""Representation of a Vue Sensor's current power."""
def __init__(self, coordinator, id):
def __init__(self, coordinator, identifier):
"""Pass coordinator to CoordinatorEntity."""
super().__init__(coordinator)
self._id = id
self._scale = coordinator.data[id]["scale"]
device_gid = coordinator.data[id]["device_gid"]
channel_num = coordinator.data[id]["channel_num"]
self._device = coordinator.data[id]["info"]
self._id = identifier
self._scale = coordinator.data[identifier]["scale"]
device_gid = coordinator.data[identifier]["device_gid"]
channel_num = coordinator.data[identifier]["channel_num"]
self._device = coordinator.data[identifier]["info"]
self._channel = None
if self._device is not None:
for channel in self._device.channels:
@@ -67,70 +69,61 @@ class CurrentVuePowerSensor(CoordinatorEntity, SensorEntity):
self._channel = channel
break
if self._channel is None:
_LOGGER.warn(
f"No channel found for device_gid {device_gid} and channel_num {channel_num}"
_LOGGER.warning(
"No channel found for device_gid %s and channel_num %s",
device_gid,
channel_num,
)
raise RuntimeError(
f"No channel found for device_gid {device_gid} and channel_num {channel_num}"
)
dName = self._device.device_name
if self._channel.name and self._channel.name not in ["Main", "Balance"]:
dName = self._channel.name
self._name = f"{dName} {channel_num} {self._scale}"
device_name = self._device.device_name
if self._channel.name and self._channel.name not in [
"Main",
"Balance",
"TotalUsage",
"MainsToGrid",
"MainsFromGrid",
]:
device_name = self._channel.name
self._name = f"{device_name} {channel_num} {self._scale}"
self._iskwh = self.scale_is_energy()
@property
def name(self):
"""Return the name of the sensor."""
return self._name
self._attr_name = self._name
if self._iskwh:
self._attr_native_unit_of_measurement = ENERGY_KILO_WATT_HOUR
self._attr_device_class = SensorDeviceClass.ENERGY
self._attr_state_class = SensorStateClass.TOTAL
else:
self._attr_native_unit_of_measurement = POWER_WATT
self._attr_device_class = SensorDeviceClass.POWER
self._attr_state_class = SensorStateClass.MEASUREMENT
@property
def state(self):
def native_value(self):
"""Return the state of the sensor."""
usage = self.coordinator.data[self._id]["usage"]
return self.scale_usage(usage)
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
if self._iskwh:
return ENERGY_KILO_WATT_HOUR
else:
return POWER_WATT
@property
def device_class(self):
"""The type of sensor"""
if self._iskwh:
return DEVICE_CLASS_ENERGY
else:
return DEVICE_CLASS_POWER
@property
def state_class(self):
"""Type of state."""
if self._iskwh:
return STATE_CLASS_TOTAL_INCREASING
else:
return STATE_CLASS_MEASUREMENT
if self._id in self.coordinator.data:
usage = self.coordinator.data[self._id]["usage"]
return self.scale_usage(usage)
return None
@property
def last_reset(self):
"""The time when the daily/monthly sensor was reset. Midnight local time."""
return self.coordinator.data[self._id]["reset"]
if self._id in self.coordinator.data:
return self.coordinator.data[self._id]["reset"]
return None
@property
def unique_id(self):
"""Unique ID for the sensor"""
if self._scale == Scale.MINUTE.value:
return f"sensor.emporia_vue.instant.{self._channel.device_gid}-{self._channel.channel_num}"
else:
return f"sensor.emporia_vue.{self._scale}.{self._channel.device_gid}-{self._channel.channel_num}"
return f"sensor.emporia_vue.{self._scale}.{self._channel.device_gid}-{self._channel.channel_num}"
@property
def device_info(self):
dName = self._channel.name or self._device.device_name
device_name = self._channel.name or self._device.device_name
return {
"identifiers": {
# Serial numbers are unique identifiers within a specific domain
@@ -141,9 +134,10 @@ class CurrentVuePowerSensor(CoordinatorEntity, SensorEntity):
),
)
},
"name": dName,
"name": device_name,
"model": self._device.model,
"sw_version": self._device.firmware,
"manufacturer": "Emporia"
# "via_device": self._device.device_gid # might be able to map the extender, nested outlets
}
@@ -163,8 +157,8 @@ class CurrentVuePowerSensor(CoordinatorEntity, SensorEntity):
def scale_is_energy(self):
"""Returns True if the scale is an energy unit instead of power (hour and bigger)"""
return (
self._scale != Scale.MINUTE.value
and self._scale != Scale.SECOND.value
and self._scale != Scale.MINUTES_15.value
return self._scale not in (
Scale.MINUTE.value,
Scale.SECOND.value,
Scale.MINUTES_15.value,
)

36
custom_components/emporia_vue/services.yaml Normal file
View File

@@ -0,0 +1,36 @@
set_charger_current:
# Service name as shown in UI
name: Set Charger Current
# Description of the service
description: Sets the charging current for an EVSE/Charger.
# If the service accepts entity IDs, target allows the user to specify entities by entity, device, or area. If `target` is specified, `entity_id` should not be defined in the `fields` map. By default it shows only targets matching entities from the same domain as the service, but if further customization is required, target supports the entity, device, and area selectors (https://www.home-assistant.io/docs/blueprint/selectors/). Entity selector parameters will automatically be applied to device and area, and device selector parameters will automatically be applied to area.
target:
entity:
manufacturer: Emporia
device_class: outlet
multiple: false
device:
manufacturer: Emporia
model: VVDN01
multiple: false
# Different fields that your service accepts
fields:
# Key of the field
current:
# Field name as shown in UI
name: Charging Current
# Description of the field
description: The desired charging current in Amps.
# Whether or not field is required (default = false)
required: true
# Advanced fields are only shown when the advanced mode is enabled for the user (default = false)
advanced: false
# Example value that can be passed for this field
example: 6
# The default field value
default: 6
# Selector (https://www.home-assistant.io/docs/blueprint/selectors/) to control the input UI for this field
selector:
number:
min: 6
max: 48

144
custom_components/emporia_vue/switch.py
View File

@@ -1,36 +1,48 @@
"""Platform for switch integration."""
import asyncio
from datetime import timedelta
import logging
import asyncio
import async_timeout
from homeassistant.components.switch import SwitchEntity
from homeassistant.components.switch import (
SwitchDeviceClass,
SwitchEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from .const import DOMAIN, VUE_DATA
from pyemvue.device import ChargerDevice, OutletDevice, VueDevice
from pyemvue import pyemvue
from pyemvue.device import OutletDevice
from .charger_entity import EmporiaChargerEntity
from .const import DOMAIN, VUE_DATA
_LOGGER = logging.getLogger(__name__)
device_information = {} # data is the populated device objects
device_information: dict[int, VueDevice] = {} # data is the populated device objects
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
):
"""Set up the sensor platform."""
vue = hass.data[DOMAIN][config_entry.entry_id][VUE_DATA]
loop = asyncio.get_event_loop()
devices = await loop.run_in_executor(None, vue.get_devices)
devices: list[VueDevice] = await loop.run_in_executor(None, vue.get_devices)
for device in devices:
if device.outlet is not None:
await loop.run_in_executor(None, vue.populate_device_properties, device)
device_information[device.device_gid] = device
elif device.ev_charger is not None:
await loop.run_in_executor(None, vue.populate_device_properties, device)
device_information[device.device_gid] = device
async def async_update_data():
"""Fetch data from API endpoint.
@@ -43,41 +55,63 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
# handled by the data update coordinator.
data = {}
loop = asyncio.get_event_loop()
outlets = await loop.run_in_executor(None, vue.get_outlets)
outlets: list[OutletDevice]
chargers: list[ChargerDevice]
(outlets, chargers) = await loop.run_in_executor(
None, vue.get_devices_status
)
if outlets:
for outlet in outlets:
data[outlet.device_gid] = outlet
if chargers:
for charger in chargers:
data[charger.device_gid] = charger
return data
except Exception as err:
raise UpdateFailed(f'Error communicating with Emporia API: {err}')
raise UpdateFailed(f"Error communicating with Emporia API: {err}")
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
# Name of the data. For logging purposes.
name='switch',
name="switch",
update_method=async_update_data,
# Polling interval. Will only be polled if there are subscribers.
update_interval=timedelta(minutes=5),
update_interval=timedelta(minutes=1),
)
await coordinator.async_refresh()
async_add_entities(
EmporiaOutletSwitch(coordinator, vue, id) for idx, id in enumerate(coordinator.data)
)
switches = []
for _, gid in enumerate(coordinator.data):
if device_information[gid].outlet:
switches.append(EmporiaOutletSwitch(coordinator, vue, gid))
elif device_information[gid].ev_charger:
switches.append(
EmporiaChargerSwitch(
coordinator,
vue,
device_information[gid],
None,
SwitchDeviceClass.OUTLET,
)
)
async_add_entities(switches)
class EmporiaOutletSwitch(CoordinatorEntity, SwitchEntity):
"""Representation of an Emporia Smart Outlet state"""
def __init__(self, coordinator, vue, id):
def __init__(self, coordinator, vue, gid):
"""Pass coordinator to CoordinatorEntity."""
super().__init__(coordinator)
#self._state = coordinator.data[index]['usage']
# self._state = coordinator.data[index]['usage']
self._vue = vue
self._device_gid = id
self._device = device_information[id]
self._name = f'Switch {self._device.device_name}'
self._device_gid = gid
self._device = device_information[gid]
self._name = f"Switch {self._device.device_name}"
self._attr_device_class = SwitchDeviceClass.OUTLET
@property
def name(self):
@@ -89,47 +123,71 @@ class EmporiaOutletSwitch(CoordinatorEntity, SwitchEntity):
"""Return the state of the switch."""
return self.coordinator.data[self._device_gid].outlet_on
# @property
# def current_power_w(self):
# """Return the current power consumption of the switch."""
# return None # so this one is sorta funny because there are separate energy sensors
# @property
# def today_energy_kwh(self):
# """Return the power consumption today for the switch."""
# return None # so this one is sorta funny because there are separate energy sensors
# @property
# def is_standby(self):
# """Indicate if the device connected to the switch is currently in standby."""
# return None # Could apply a semi-arbitrary limit of like 5 watts for this
async def async_turn_on(self, **kwargs):
"""Turn the switch on."""
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, self._vue.update_outlet, self.coordinator.data[self._device_gid], True)
await loop.run_in_executor(
None, self._vue.update_outlet, self.coordinator.data[self._device_gid], True
)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs):
"""Turn the switch off."""
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, self._vue.update_outlet, self.coordinator.data[self._device_gid], False)
await loop.run_in_executor(
None,
self._vue.update_outlet,
self.coordinator.data[self._device_gid],
False,
)
await self.coordinator.async_request_refresh()
@property
def unique_id(self):
"""Unique ID for the switch"""
return f'switch.emporia_vue.{self._device_gid}'
return f"switch.emporia_vue.{self._device_gid}"
@property
def device_info(self):
return {
"identifiers": {
# Serial numbers are unique identifiers within a specific domain
(DOMAIN, '{0}-1,2,3'.format(self._device_gid))
(DOMAIN, "{0}-1,2,3".format(self._device_gid))
},
"name": self._device.device_name+'-1,2,3',
"name": self._device.device_name + "-1,2,3",
"model": self._device.model,
"sw_version": self._device.firmware,
#"via_device": self._device.device_gid # might be able to map the extender, nested outlets
"manufacturer": "Emporia"
# "via_device": self._device.device_gid # might be able to map the extender, nested outlets
}
class EmporiaChargerSwitch(EmporiaChargerEntity, SwitchEntity):
"""Representation of an Emporia Charger switch state"""
@property
def is_on(self):
"""Return the state of the switch."""
return self.coordinator.data[self._device.device_gid].charger_on
async def async_turn_on(self, **kwargs):
"""Turn the charger on."""
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None,
self._vue.update_charger,
self._coordinator.data[self._device.device_gid],
True,
)
await self._coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs):
"""Turn the charger off."""
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None,
self._vue.update_charger,
self._coordinator.data[self._device.device_gid],
False,
)
await self._coordinator.async_request_refresh()