diff --git a/SunGather/config-example.yaml b/SunGather/config-example.yaml index 2484180..a019e75 100644 --- a/SunGather/config-example.yaml +++ b/SunGather/config-example.yaml @@ -155,3 +155,18 @@ exports: # register: # - name: "m1" # Text Message 1 - Donation Only # register: + + # Push data to ChargeHQ for EV charging optimization + # See: https://chargehq.net/kb/push-api + - name: chargehq + enabled: False # [Optional] Default is False + api_key: "xxxxx" # [Required] Get from ChargeHQ app: My Equipment > Solar/Battery Equipment > Push API + # push_interval: 60 # [Optional] Default 60, minimum 30 seconds between pushes + # Register mappings - defaults work for hybrid inverters (SH* models) + # For grid-tied inverters (SG* models) you may need to adjust these: + # register_production: total_active_power # Solar production in Watts + # register_consumption: load_power_hybrid # Load consumption in Watts (use load_power for SG* models) + # register_net_import: export_power_hybrid # Grid import/export in Watts (use meter_power for SG* models) + # register_battery_power: battery_power # Battery charge/discharge in Watts (hybrid only) + # register_battery_soc: battery_level # Battery state of charge in % (hybrid only) + # invert_net_import: True # Set False if using meter_power (positive=import) diff --git a/SunGather/exports/chargehq.py b/SunGather/exports/chargehq.py new file mode 100644 index 0000000..be7a105 --- /dev/null +++ b/SunGather/exports/chargehq.py @@ -0,0 +1,126 @@ +import logging +import requests +import time + +class export_chargehq(object): + """ + ChargeHQ Push API export + See: https://chargehq.net/kb/push-api + + Pushes solar/battery data to ChargeHQ for EV charging optimization. + Rate limit: minimum 30 seconds between calls (60 seconds recommended). + """ + + def __init__(self): + self.api_url = "https://api.chargehq.net/api/public/push-solar-data" + self.last_push = 0 + + def configure(self, config, inverter): + self.api_key = config.get('api_key') + if not self.api_key: + logging.error("ChargeHQ: api_key is required") + return False + + # Push interval in seconds (minimum 30, default 60) + self.push_interval = max(30, config.get('push_interval', 60)) + + # Register mappings - users can override these in config + # Defaults work for most Sungrow hybrid inverters + self.reg_production = config.get('register_production', 'total_active_power') + self.reg_consumption = config.get('register_consumption', 'load_power_hybrid') + self.reg_net_import = config.get('register_net_import', 'export_power_hybrid') + self.reg_battery_power = config.get('register_battery_power', 'battery_power') + self.reg_battery_soc = config.get('register_battery_soc', 'battery_level') + + # Whether net_import register is inverted (export_power_hybrid is positive when exporting) + self.invert_net_import = config.get('invert_net_import', True) + + logging.info(f"ChargeHQ: Configured with push interval {self.push_interval}s") + return True + + def publish(self, inverter): + # Rate limiting + now = time.time() + if (now - self.last_push) < self.push_interval: + remaining = int(self.push_interval - (now - self.last_push)) + logging.debug(f"ChargeHQ: Waiting {remaining}s before next push") + return True + + payload = { + "apiKey": self.api_key, + "siteMeters": {} + } + + site_meters = payload["siteMeters"] + + # Production (solar generation) - convert W to kW + production = self._get_register_value(inverter, self.reg_production) + if production is not None: + site_meters["production_kw"] = round(production / 1000, 3) + + # Consumption (load power) - convert W to kW + consumption = self._get_register_value(inverter, self.reg_consumption) + if consumption is not None: + site_meters["consumption_kw"] = round(abs(consumption) / 1000, 3) + + # Net import (positive = importing, negative = exporting) - convert W to kW + net_import = self._get_register_value(inverter, self.reg_net_import) + if net_import is not None: + if self.invert_net_import: + net_import = -net_import # export_power_hybrid is positive when exporting + site_meters["net_import_kw"] = round(net_import / 1000, 3) + + # Battery discharge (positive = discharging, negative = charging) - convert W to kW + battery_power = self._get_register_value(inverter, self.reg_battery_power) + if battery_power is not None: + site_meters["battery_discharge_kw"] = round(battery_power / 1000, 3) + + # Battery SOC (ChargeHQ expects 0-1, battery_level is 0-100%) + battery_soc = self._get_register_value(inverter, self.reg_battery_soc) + if battery_soc is not None: + site_meters["battery_soc"] = round(battery_soc / 100, 3) + + # Only push if we have at least production data + if "production_kw" not in site_meters: + logging.warning("ChargeHQ: No production data available, skipping push") + return False + + try: + logging.debug(f"ChargeHQ: Pushing data: {site_meters}") + response = requests.post( + self.api_url, + json=payload, + headers={"Content-Type": "application/json"}, + timeout=10 + ) + + if response.status_code == 200: + self.last_push = now + logging.info(f"ChargeHQ: Data pushed successfully - production={site_meters.get('production_kw', 0)}kW") + return True + else: + logging.error(f"ChargeHQ: Push failed with status {response.status_code}: {response.text}") + return False + + except requests.exceptions.Timeout: + logging.error("ChargeHQ: Request timed out") + return False + except requests.exceptions.RequestException as err: + logging.error(f"ChargeHQ: Request failed: {err}") + return False + + def _get_register_value(self, inverter, register_name): + """Get a register value if it exists in the latest scrape.""" + if not register_name: + return None + if not inverter.validateLatestScrape(register_name): + logging.debug(f"ChargeHQ: Register {register_name} not in latest scrape") + return None + value = inverter.getRegisterValue(register_name) + if value is None: + return None + try: + return float(value) + except (ValueError, TypeError): + logging.warning(f"ChargeHQ: Could not convert {register_name}={value} to float") + return None