Created
October 7, 2025 18:05
-
-
Save markfickett/04b01d5427b06a2dcdfbaad12179c061 to your computer and use it in GitHub Desktop.
AppDaemon app to connect vuegraf and OpenEVSE via MQTT
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """Forwards Emporia Vue Watts value to OpenEVSE for the Solar Divert feature. | |
| This connects an Emporia Vue's net usage reading to the OpenEVSE eco charging | |
| mode. | |
| The Vue needs vuegraf (https://github.com/jertel/vuegraf) to publish its data | |
| to MQTT. (The Emporia API provides fresh values every 60s at most often.) | |
| The OpenEVSE needs the "Solar Divert" service turned on and configured to read | |
| from this plugin's "publish_net_watts_topic" MQTT topic. Then switch to | |
| "Eco (PV Direct)" in the OpenEVSE Charge Options. | |
| This plugin then reads the usage value from vuegraf and writes it as expected | |
| for the OpenEVSE. To see the published values, use for example: | |
| mosquitto.sub -u <user> -P <pw> -t 'vuegraf/energy_usage/net' -v | |
| Example app config: | |
| evse_surplus_solar: | |
| module: evse_surplus_solar | |
| class: EvseSurplusSolar | |
| enabled_switch: input_boolean.evse_surplus_only | |
| vuegraf: | |
| usage_topic: vuegraf/energy_usage | |
| account: "Left Panel" | |
| net_device: "Vue Gen3 Unit 3" | |
| publish_net_watts_topic: vuegraf/energy_usage/net | |
| openevse: | |
| base_topic: openevse | |
| """ | |
| from dataclasses import dataclass | |
| import enum | |
| import json | |
| from typing import Dict | |
| import hassapi as hass | |
| _MQTT_PLUGIN_NAMESPACE = "mqtt" # matches appdeamon.yaml | |
| # OpenEVSE divertmode modes, from | |
| # https://github.com/OpenEVSE/openevse_esp32_firmware/blob/master/src/mqtt.cpp#L330 | |
| _DIVERT_ENABLED = "2" | |
| _DIVERT_DISABLED = "1" | |
| class EvseSurplusSolar(hass.Hass): | |
| def initialize(self) -> None: | |
| self._mqtt = self.get_plugin_api("MQTT") | |
| self._enabled_switch = self.args["enabled_switch"] | |
| self.listen_state(self._on_enabled_change, self._enabled_switch) | |
| self._enabled = self.get_state(self._enabled_switch) == "on" | |
| # Input topic containing net usage (and other fields). | |
| self._vuegraf_topic = self.args["vuegraf"]["usage_topic"] | |
| # Expected account and device. Only messages from this device are used. | |
| self._vuegraf_account = self.args["vuegraf"]["account"] | |
| self._vuegraf_net_device = self.args["vuegraf"]["net_device"] | |
| # Output topic for just the net usage watts value. | |
| self._vuegraf_net_watts_topic = self.args["vuegraf"]["publish_net_watts_topic"] | |
| openevse_base = self.args["openevse"]["base_topic"] | |
| assert not openevse_base.endswith("/") | |
| self._openevse_divertmode_topic = openevse_base + "/divertmode" | |
| self._openevse_divertmode_set_topic = openevse_base + "/divertmode/set" | |
| for (topic, listener) in ( | |
| (self._vuegraf_topic, self._on_energy_usage_received), | |
| (self._openevse_divertmode_topic, self._on_openevse_divertmode_received), | |
| ): | |
| self._mqtt.mqtt_subscribe(topic, namespace=_MQTT_PLUGIN_NAMESPACE) | |
| self._mqtt.listen_event(listener, "MQTT_MESSAGE", topic=topic) | |
| def _on_energy_usage_received( | |
| self, event: str, data: Dict, unused_kwargs: Dict | |
| ) -> None: | |
| """Unpacks an MQTT message from vuegraf and forwards just the net usage (W). | |
| Example input: | |
| vuegraf/energy_usage {"account": "67 Church", "device_name": | |
| "Vue Gen3 Unit 2", "usage_watts": 1404.500801340739, | |
| "epoch_s": 1759495980, "detailed": "False"} | |
| Example output: | |
| vuegraf/energy_usage/net 1404 | |
| """ | |
| usage = json.loads(data["payload"]) | |
| if ( | |
| usage["account"] == self._vuegraf_account | |
| and usage["device_name"] == self._vuegraf_net_device | |
| ): | |
| usage = int(float(usage["usage_watts"])) | |
| if self._enabled: | |
| self.log(f"Updating for {self._vuegraf_net_device} at {usage}W.") | |
| self._mqtt.mqtt_publish( | |
| self._vuegraf_net_watts_topic, | |
| payload=str(usage), | |
| namespace=_MQTT_PLUGIN_NAMESPACE, | |
| ) | |
| def _on_enabled_change(self, entity, attribute, old, new, kwargs): | |
| """Mirrors the HA input bool to the OpenEVSE Charge Option (normal v. eco).""" | |
| enabled = (new == "on") | |
| if enabled != self._enabled: | |
| self._enabled = enabled | |
| self.log(f"Changed to {self._enabled=} from HomeAssistant") | |
| self._mqtt.mqtt_publish( | |
| self._openevse_divertmode_set_topic, | |
| payload=_DIVERT_ENABLED if self._enabled else _DIVERT_DISABLED, | |
| namespace=_MQTT_PLUGIN_NAMESPACE, | |
| ) | |
| def _on_openevse_divertmode_received( | |
| self, event: str, data: Dict, unused_kwargs: Dict | |
| ) -> None: | |
| """Mirrors the OpenEVSE Charge Option (normal v. eco) to the HA input bool.""" | |
| mode = data["payload"] | |
| if mode == _DIVERT_DISABLED and self._enabled: | |
| self._enabled = False | |
| self.log(f"OpenEVSE divertmode changed to {mode=} {self._enabled=}") | |
| self.turn_off(self._enabled_switch) | |
| elif mode == _DIVERT_ENABLED and not self._enabled: | |
| self._enabled = True | |
| self.log(f"OpenEVSE divertmode changed to {mode=} {self._enabled=}") | |
| self.turn_on(self._enabled_switch) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment