From 4fcf40408f4321ff8245d6332ce140b6e68004d2 Mon Sep 17 00:00:00 2001 From: Dan Raper Date: Sat, 30 Dec 2023 16:44:39 +0000 Subject: [PATCH] Added voltage and next slot end sensor, and charge slot active binary sensor (#20) * Add CT detection * Add next slot ends sensor * Added voltage sensor * Added charge slot active sensor --- README.md | 14 +-- custom_components/ohme/api_client.py | 9 ++ custom_components/ohme/binary_sensor.py | 72 +++++++++++++-- custom_components/ohme/sensor.py | 114 ++++++++++++++++++++++-- custom_components/ohme/utils.py | 52 ++++++++--- 5 files changed, 228 insertions(+), 33 deletions(-) diff --git a/README.md b/README.md index 82fcde3..07f51a9 100644 --- a/README.md +++ b/README.md @@ -37,17 +37,21 @@ This integration exposes the following entities: * Car Connected - On when a car is plugged in * Car Charging - On when a car is connected and drawing power * Pending Approval - On when a car is connected and waiting for approval -* Sensors + * Charge Slot Active - On when a charge slot is in progress according to the Ohme-generated charge plan +* Sensors (Charge power) - **These are only available during a charge session** * Power Draw (Watts) - Power draw of connected car * Current Draw (Amps) - Current draw of connected car + * Voltage (Volts) - Voltage reading +* Sensors (Other) * CT Reading (Amps) - Reading from attached CT clamp * Accumulative Energy Usage (kWh) - Total energy used by the charger - * Next Smart Charge Slot - The next time your car will start charging according to the Ohme-generated charge plan -* Switches (Settings) - Only options available to your charger model will show + * Next Charge Slot Start - The next time your car will start charging according to the Ohme-generated charge plan + * Next Charge Slot End - The next time your car will stop charging according to the Ohme-generated charge plan +* Switches (Settings) - **Only options available to your charger model will show** * Lock Buttons - Locks buttons on charger * Require Approval - Require approval to start a charge * Sleep When Inactive - Charger screen & lights will automatically turn off -* Switches (Charge state) - These are only functional when a car is connected +* Switches (Charge state) - **These are only functional when a car is connected** * Max Charge - Forces the connected car to charge regardless of set schedule * Pause Charge - Pauses an ongoing charge * Buttons @@ -61,7 +65,7 @@ The coordinators are listed with their refresh intervals below. Relevant coordin * OhmeChargeSessionsCoordinator (30s refresh) * Binary Sensors: All * Buttons: Approve Charge - * Sensors: Power, current and next slot + * Sensors: Power, current, voltage and next slot (start & end) * Switches: Max charge, pause charge * OhmeAccountInfoCoordinator (1m refresh) * Switches: Lock buttons, require approval and sleep when inactive diff --git a/custom_components/ohme/api_client.py b/custom_components/ohme/api_client.py index b0c2162..53451b5 100644 --- a/custom_components/ohme/api_client.py +++ b/custom_components/ohme/api_client.py @@ -25,6 +25,7 @@ def __init__(self, email, password): # Charger and its capabilities self._device_info = None self._capabilities = {} + self._ct_connected = False # Authentication self._token_birth = 0 @@ -149,6 +150,10 @@ async def _get_request(self, url): # Simple getters + def ct_connected(self): + """Is CT clamp connected.""" + return self._ct_connected + def is_capable(self, capability): """Return whether or not this model has a given capability.""" return bool(self._capabilities[capability]) @@ -241,6 +246,10 @@ async def async_get_charge_statistics(self): async def async_get_ct_reading(self): """Get CT clamp reading.""" resp = await self._get_request(f"/v1/chargeDevices/{self._serial}/advancedSettings") + + # If we ever get a reading above 0, assume CT connected + if resp['clampAmps'] > 0: + self._ct_connected = True return resp['clampAmps'] diff --git a/custom_components/ohme/binary_sensor.py b/custom_components/ohme/binary_sensor.py index e41f90f..d8bad2b 100644 --- a/custom_components/ohme/binary_sensor.py +++ b/custom_components/ohme/binary_sensor.py @@ -6,10 +6,12 @@ BinarySensorEntity ) from homeassistant.helpers.update_coordinator import CoordinatorEntity -from homeassistant.core import HomeAssistant +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity import generate_entity_id +from homeassistant.util.dt import (utcnow) from .const import DOMAIN, DATA_COORDINATORS, COORDINATOR_CHARGESESSIONS, DATA_CLIENT from .coordinator import OhmeChargeSessionsCoordinator +from .utils import charge_graph_in_slot async def async_setup_entry( @@ -21,14 +23,15 @@ async def async_setup_entry( client = hass.data[DOMAIN][DATA_CLIENT] coordinator = hass.data[DOMAIN][DATA_COORDINATORS][COORDINATOR_CHARGESESSIONS] - sensors = [ConnectedSensor(coordinator, hass, client), - ChargingSensor(coordinator, hass, client), - PendingApprovalSensor(coordinator, hass, client)] + sensors = [ConnectedBinarySensor(coordinator, hass, client), + ChargingBinarySensor(coordinator, hass, client), + PendingApprovalBinarySensor(coordinator, hass, client), + CurrentSlotBinarySensor(coordinator, hass, client)] async_add_entities(sensors, update_before_add=True) -class ConnectedSensor( +class ConnectedBinarySensor( CoordinatorEntity[OhmeChargeSessionsCoordinator], BinarySensorEntity): """Binary sensor for if car is plugged in.""" @@ -74,7 +77,7 @@ def is_on(self) -> bool: return self._state -class ChargingSensor( +class ChargingBinarySensor( CoordinatorEntity[OhmeChargeSessionsCoordinator], BinarySensorEntity): """Binary sensor for if car is charging.""" @@ -121,7 +124,7 @@ def is_on(self) -> bool: return self._state -class PendingApprovalSensor( +class PendingApprovalBinarySensor( CoordinatorEntity[OhmeChargeSessionsCoordinator], BinarySensorEntity): """Binary sensor for if a charge is pending approval.""" @@ -165,3 +168,58 @@ def is_on(self) -> bool: self.coordinator.data["mode"] == "PENDING_APPROVAL") return self._state + + +class CurrentSlotBinarySensor( + CoordinatorEntity[OhmeChargeSessionsCoordinator], + BinarySensorEntity): + """Binary sensor for if we are currently in a smart charge slot.""" + + _attr_name = "Charge Slot Active" + + def __init__( + self, + coordinator: OhmeChargeSessionsCoordinator, + hass: HomeAssistant, + client): + super().__init__(coordinator=coordinator) + + self._attributes = {} + self._last_updated = None + self._state = False + self._client = client + + self.entity_id = generate_entity_id( + "binary_sensor.{}", "ohme_slot_active", hass=hass) + + self._attr_device_info = hass.data[DOMAIN][DATA_CLIENT].get_device_info( + ) + + @property + def icon(self): + """Icon of the sensor.""" + return "mdi:calendar-check" + + @property + def unique_id(self) -> str: + """Return the unique ID of the sensor.""" + return self._client.get_unique_id("ohme_slot_active") + + @property + def is_on(self) -> bool: + return self._state + + @callback + def _handle_coordinator_update(self) -> None: + """Are we in a charge slot? This is a bit slow so we only update on coordinator data update.""" + if self.coordinator.data is None: + self._state = None + elif self.coordinator.data["mode"] == "DISCONNECTED": + self._state = False + else: + self._state = charge_graph_in_slot( + self.coordinator.data['startTime'], self.coordinator.data['chargeGraph']['points']) + + self._last_updated = utcnow() + + self.async_write_ha_state() diff --git a/custom_components/ohme/sensor.py b/custom_components/ohme/sensor.py index 1880e52..dd64c3b 100644 --- a/custom_components/ohme/sensor.py +++ b/custom_components/ohme/sensor.py @@ -6,7 +6,7 @@ SensorEntity ) from homeassistant.helpers.update_coordinator import CoordinatorEntity -from homeassistant.const import UnitOfPower, UnitOfEnergy, UnitOfElectricCurrent +from homeassistant.const import UnitOfPower, UnitOfEnergy, UnitOfElectricCurrent, UnitOfElectricPotential from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity import generate_entity_id from homeassistant.util.dt import (utcnow) @@ -30,9 +30,11 @@ async def async_setup_entry( sensors = [PowerDrawSensor(coordinator, hass, client), CurrentDrawSensor(coordinator, hass, client), + VoltageSensor(coordinator, hass, client), CTSensor(adv_coordinator, hass, client), EnergyUsageSensor(stats_coordinator, hass, client), - NextSlotSensor(coordinator, hass, client)] + NextSlotEndSensor(coordinator, hass, client), + NextSlotStartSensor(coordinator, hass, client)] async_add_entities(sensors, update_before_add=True) @@ -121,7 +123,49 @@ def native_value(self): return 0 -class CTSensor(CoordinatorEntity[OhmeChargeSessionsCoordinator], SensorEntity): +class VoltageSensor(CoordinatorEntity[OhmeChargeSessionsCoordinator], SensorEntity): + """Sensor for EVSE voltage.""" + _attr_name = "Voltage" + _attr_device_class = SensorDeviceClass.VOLTAGE + _attr_native_unit_of_measurement = UnitOfElectricPotential.VOLT + + def __init__( + self, + coordinator: OhmeChargeSessionsCoordinator, + hass: HomeAssistant, + client): + super().__init__(coordinator=coordinator) + + self._state = None + self._attributes = {} + self._last_updated = None + self._client = client + + self.entity_id = generate_entity_id( + "sensor.{}", "ohme_voltage", hass=hass) + + self._attr_device_info = hass.data[DOMAIN][DATA_CLIENT].get_device_info( + ) + + @property + def unique_id(self) -> str: + """Return the unique ID of the sensor.""" + return self._client.get_unique_id("voltage") + + @property + def icon(self): + """Icon of the sensor.""" + return "mdi:sine-wave" + + @property + def native_value(self): + """Get value from data returned from API by coordinator""" + if self.coordinator.data and self.coordinator.data['power']: + return self.coordinator.data['power']['volt'] + return None + + +class CTSensor(CoordinatorEntity[OhmeAdvancedSettingsCoordinator], SensorEntity): """Sensor for car power draw.""" _attr_name = "CT Reading" _attr_device_class = SensorDeviceClass.CURRENT @@ -129,7 +173,7 @@ class CTSensor(CoordinatorEntity[OhmeChargeSessionsCoordinator], SensorEntity): def __init__( self, - coordinator: OhmeChargeSessionsCoordinator, + coordinator: OhmeAdvancedSettingsCoordinator, hass: HomeAssistant, client): super().__init__(coordinator=coordinator) @@ -171,7 +215,7 @@ class EnergyUsageSensor(CoordinatorEntity[OhmeStatisticsCoordinator], SensorEnti def __init__( self, - coordinator: OhmeChargeSessionsCoordinator, + coordinator: OhmeStatisticsCoordinator, hass: HomeAssistant, client): super().__init__(coordinator=coordinator) @@ -206,9 +250,9 @@ def native_value(self): return None -class NextSlotSensor(CoordinatorEntity[OhmeStatisticsCoordinator], SensorEntity): - """Sensor for next smart charge slot.""" - _attr_name = "Next Smart Charge Slot" +class NextSlotStartSensor(CoordinatorEntity[OhmeChargeSessionsCoordinator], SensorEntity): + """Sensor for next smart charge slot start time.""" + _attr_name = "Next Charge Slot Start" _attr_device_class = SensorDeviceClass.TIMESTAMP def __init__( @@ -234,6 +278,58 @@ def unique_id(self) -> str: """Return the unique ID of the sensor.""" return self._client.get_unique_id("next_slot") + @property + def icon(self): + """Icon of the sensor.""" + return "mdi:clock-star-four-points" + + @property + def native_value(self): + """Return pre-calculated state.""" + return self._state + + @callback + def _handle_coordinator_update(self) -> None: + """Calculate next timeslot. This is a bit slow so we only update on coordinator data update.""" + if self.coordinator.data is None or self.coordinator.data["mode"] == "DISCONNECTED": + self._state = None + else: + self._state = charge_graph_next_slot( + self.coordinator.data['startTime'], self.coordinator.data['chargeGraph']['points'])['start'] + + self._last_updated = utcnow() + + self.async_write_ha_state() + + +class NextSlotEndSensor(CoordinatorEntity[OhmeChargeSessionsCoordinator], SensorEntity): + """Sensor for next smart charge slot end time.""" + _attr_name = "Next Charge Slot End" + _attr_device_class = SensorDeviceClass.TIMESTAMP + + def __init__( + self, + coordinator: OhmeChargeSessionsCoordinator, + hass: HomeAssistant, + client): + super().__init__(coordinator=coordinator) + + self._state = None + self._attributes = {} + self._last_updated = None + self._client = client + + self.entity_id = generate_entity_id( + "sensor.{}", "ohme_next_slot_end", hass=hass) + + self._attr_device_info = hass.data[DOMAIN][DATA_CLIENT].get_device_info( + ) + + @property + def unique_id(self) -> str: + """Return the unique ID of the sensor.""" + return self._client.get_unique_id("next_slot_end") + @property def icon(self): """Icon of the sensor.""" @@ -251,7 +347,7 @@ def _handle_coordinator_update(self) -> None: self._state = None else: self._state = charge_graph_next_slot( - self.coordinator.data['startTime'], self.coordinator.data['chargeGraph']['points']) + self.coordinator.data['startTime'], self.coordinator.data['chargeGraph']['points'])['end'] self._last_updated = utcnow() diff --git a/custom_components/ohme/utils.py b/custom_components/ohme/utils.py index 7b209e3..06eed23 100644 --- a/custom_components/ohme/utils.py +++ b/custom_components/ohme/utils.py @@ -3,23 +3,27 @@ import pytz -def charge_graph_next_slot(charge_start, points): - """Get the next charge slot from a list of graph points.""" - # Get start and current timestamp in seconds +def _format_charge_graph(charge_start, points): + """Convert relative time in points array to real timestamp (s).""" + charge_start = round(charge_start / 1000) - now = int(time()) + return [{"t": x["x"] + charge_start, "y": x["y"]} for x in points] - # Replace relative timestamp (seconds) with real timestamp - data = [{"t": x["x"] + charge_start, "y": x["y"]} for x in points] + +def charge_graph_next_slot(charge_start, points): + """Get the next charge slot start/end times from a list of graph points.""" + now = int(time()) + data = _format_charge_graph(charge_start, points) # Filter to points from now onwards data = [x for x in data if x["t"] > now] # Give up if we have less than 3 points if len(data) < 3: - return None + return {"start": None, "end": None} - next_ts = None + start_ts = None + end_ts = None # Loop through every remaining value, skipping the last for idx in range(0, len(data) - 1): @@ -28,13 +32,37 @@ def charge_graph_next_slot(charge_start, points): # If the next point has a Y delta of 10+, consider this the start of a slot # This should be 0+ but I had some strange results in testing... revisit - if delta > 10: + if delta > 10 and not start_ts: # 1s added here as it otherwise often rounds down to xx:59:59 - next_ts = data[idx]["t"] + 1 + start_ts = data[idx]["t"] + 1 + + # Take the first delta of 0 as the end + if delta == 0 and not end_ts: + end_ts = data[idx]["t"] + 1 + + # These need to be presented with tzinfo or Home Assistant will reject them + return { + "start": datetime.utcfromtimestamp(start_ts).replace(tzinfo=pytz.utc) if start_ts else None, + "end": datetime.utcfromtimestamp(end_ts).replace(tzinfo=pytz.utc) if end_ts else None, + } + + +def charge_graph_in_slot(charge_start, points): + """Are we currently in a charge slot?""" + now = int(time()) + data = _format_charge_graph(charge_start, points) + + # Loop through every value, skipping the last + for idx in range(0, len(data) - 1): + # This is our current point + if data[idx]["t"] < now and data[idx + 1]["t"] > now: + # If the delta line we are on is steeper than 10, + # we are in a charge slot. + if data[idx + 1]["y"] - data[idx]["y"] > 10: + return True break - # This needs to be presented with tzinfo or Home Assistant will reject it - return None if next_ts is None else datetime.utcfromtimestamp(next_ts).replace(tzinfo=pytz.utc) + return False def time_next_occurs(hour, minute):