From 7f7be5d99bfc357e5f741885afe2ebc84f55926f Mon Sep 17 00:00:00 2001 From: bartdw Date: Wed, 15 Oct 2025 20:59:11 +0200 Subject: [PATCH] add charger functionality --- pysmappee/api.py | 13 ++++++ pysmappee/charger.py | 86 ++++++++++++++++++++++++++++++++++++ pysmappee/config.py | 3 ++ pysmappee/mqtt.py | 7 +++ pysmappee/servicelocation.py | 29 ++++++++++++ 5 files changed, 138 insertions(+) create mode 100644 pysmappee/charger.py diff --git a/pysmappee/api.py b/pysmappee/api.py index ed3bd71..f09af53 100644 --- a/pysmappee/api.py +++ b/pysmappee/api.py @@ -207,6 +207,19 @@ def get_actuator_connection_state(self, service_location_id, actuator_id): r = requests.get(url, headers=self.headers) r.raise_for_status() return r.text + + @authenticated + def set_charging_mode(self, chargingStationSerialNumber, position, payload): + url = urljoin( + config['API_URL'][self._farm]['chargingstations_url'], + chargingStationSerialNumber, + "connectors", + position, + "mode" + ) + r = requests.put(url, headers=self.headers, json=payload) + r.raise_for_status() + return r def _to_milliseconds(self, time): if isinstance(time, dt.datetime): diff --git a/pysmappee/charger.py b/pysmappee/charger.py new file mode 100644 index 0000000..b8cad41 --- /dev/null +++ b/pysmappee/charger.py @@ -0,0 +1,86 @@ +"""Support for Charging Stations.""" + +from enum import Enum + +class ChargingMode(Enum): + NORMAL = 'NORMAL' + SMART = 'SMART' + PAUSED = 'PAUSED' + + +class SmappeeCharger: + """Representation of a Smappee Charger.""" + + def __init__(self, + service_location, + chargingStationSerialNumber, + position, + uuid, + serialNumber, + minPower=None, + maxPower=None, + minCurrent=None, + maxCurrent=None): + # configuration details + self.service_location = service_location + self.chargingStationSerialNumber = chargingStationSerialNumber + self.position = position + self.uuid = uuid + self.serialNumber = serialNumber + self.minPower = minPower + self.maxPower = maxPower + self.minCurrent = minCurrent + self.maxCurrent = maxCurrent + self.percentageLimit = None + self.chargingState = None + self.chargingMode = None + self.optimizationStrategy = None + + @property + def chargingCurrent(self): + """Get the actual current in Amperes.""" + if self.percentageLimit is not None and self.maxCurrent is not None: + return self.minCurrent + (self.maxCurrent - self.minCurrent) * self.percentageLimit / 100.0 + return None + + def set_charging_mode(self, mode, percentage=None, current=None): + """Set the charging mode of this charger.""" + # mode: "off", "min", "max", "custom" + + if mode == ChargingMode.SMART and percentage is not None: + raise ValueError("Cannot set percentage when mode is SMART") + if mode == ChargingMode.SMART and current is not None: + raise ValueError("Cannot set current when mode is SMART") + + payload = { + "mode": mode.value + } + if percentage is not None: + payload['limit'] = { + "unit": "PERCENTAGE", + "value": percentage + } + elif current is not None: + payload['limit'] = { + "unit": "AMPERE", + "value": current + } + + self.service_location.smappee_api.set_charging_mode( + self.chargingStationSerialNumber, + self.position, + payload + ) + + def update_from_mqtt(self, payload): + """Update the charger state from an MQTT message payload.""" + if 'percentageLimit' in payload: + self.percentageLimit = payload['percentageLimit'] + if 'chargingState' in payload: + self.chargingState = payload['chargingState'] + if 'chargingMode' in payload: + self.chargingMode = payload['chargingMode'] + if 'optimizationStrategy' in payload: + self.optimizationStrategy = payload['optimizationStrategy'] + + diff --git a/pysmappee/config.py b/pysmappee/config.py index f2555df..686d12c 100644 --- a/pysmappee/config.py +++ b/pysmappee/config.py @@ -6,14 +6,17 @@ 'authorize_url': 'https://app1pub.smappee.net/dev/v1/oauth2/authorize', 'token_url': 'https://app1pub.smappee.net/dev/v3/oauth2/token', 'servicelocation_url': 'https://app1pub.smappee.net/dev/v3/servicelocation', + 'chargingstations_url': 'https://app1pub.smappee.net/dev/v3/chargingstations', }, 2: { 'token_url': 'https://farm2pub.smappee.net/dev/v3/oauth2/token', 'servicelocation_url': 'https://farm2pub.smappee.net/dev/v3/servicelocation', + 'chargingstations_url': 'https://farm2pub.smappee.net/dev/v3/chargingstations', }, 3: { 'token_url': 'https://farm3pub.smappee.net/dev/v3/oauth2/token', 'servicelocation_url': 'https://farm3pub.smappee.net/dev/v3/servicelocation', + 'chargingstations_url': 'https://farm3pub.smappee.net/dev/v3/chargingstations', }, } diff --git a/pysmappee/mqtt.py b/pysmappee/mqtt.py index 721abb0..be6da65 100644 --- a/pysmappee/mqtt.py +++ b/pysmappee/mqtt.py @@ -169,6 +169,13 @@ def _on_message(self, client, userdata, message): state=plug_state, since=plug_state_since, api=False) + # carcharger topics + elif message.topic.startswith(f'{self.topic_prefix}/etc/carcharger/acchargingcontroller/v1/devices/') \ + and message.topic.endswith('/property/chargingstate'): + device_id = message.topic.split('/')[7] + charger = self._service_location.chargers[device_id] + charger.update_from_mqtt(json.loads(message.payload)) + # smart device and ETC topics elif message.topic.startswith(f'{self.topic_prefix}/etc/'): diff --git a/pysmappee/servicelocation.py b/pysmappee/servicelocation.py index c1f4669..af5989c 100644 --- a/pysmappee/servicelocation.py +++ b/pysmappee/servicelocation.py @@ -5,6 +5,7 @@ from .helper import is_smappee_solar, is_smappee_genius, is_smappee_connect, is_smappee_plus from .measurement import SmappeeMeasurement from .sensor import SmappeeSensor +from .charger import SmappeeCharger from cachetools import TTLCache @@ -41,6 +42,7 @@ def __init__(self, device_serial_number, smappee_api, service_location_id=None, self._actuators = {} self._sensors = {} self._measurements = {} + self._chargers = {} # realtime values self._realtime_values = { @@ -207,6 +209,15 @@ def load_configuration(self, refresh=False): name=sensor.get('name'), channels=sensor.get('channels')) + # Load charging stations + for cs in sl_metering_configuration.get('chargingStations', []): + chargingStationSerialNumber = cs.get('serialNumber') + + for charger in cs.get('chargers'): + self._add_charger(chargingStationSerialNumber=chargingStationSerialNumber, + rest_payload=charger + ) + # Set phase type self.phase_type = sl_metering_configuration.get('phaseType') if 'phaseType' in sl_metering_configuration else None @@ -434,6 +445,24 @@ def _add_measurement(self, id, name, type, subcircuitType, channels): type=type, subcircuit_type=subcircuitType, channels=channels) + + @property + def chargers(self): + return self._chargers + + def _add_charger(self, chargingStationSerialNumber, rest_payload): + uuid = rest_payload.get('uuid') + self.chargers[uuid] = SmappeeCharger( + service_location=self, + chargingStationSerialNumber=chargingStationSerialNumber, + position=rest_payload.get('position'), + uuid=uuid, + serialNumber=rest_payload.get('serialNumber'), + minPower=rest_payload.get('minPower'), + maxPower=rest_payload.get('maxPower'), + minCurrent=rest_payload.get('minCurrent'), + maxCurrent=rest_payload.get('maxCurrent') + ) @property def total_power(self):