diff --git a/packages/modules/devices/tesla/tesla/counter.py b/packages/modules/devices/tesla/tesla/counter.py
index 8840d2e70b..586214e9dd 100644
--- a/packages/modules/devices/tesla/tesla/counter.py
+++ b/packages/modules/devices/tesla/tesla/counter.py
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
import logging
-from requests import HTTPError
+import math
from modules.common.abstract_device import AbstractCounter
from modules.common.component_state import CounterState
@@ -18,33 +18,140 @@ def __init__(self, component_config: TeslaCounterSetup) -> None:
self.component_config = component_config
def initialize(self) -> None:
- self.store = get_component_value_store(self.component_config.type, self.component_config.id)
- self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
+ self.store = get_component_value_store(
+ self.component_config.type,
+ self.component_config.id,
+ )
+ self.fault_state = FaultState(
+ ComponentInfo.from_component_config(self.component_config)
+ )
+
+ @staticmethod
+ def _nearly_zero(x: float, eps: float = 1e-9) -> bool:
+ return abs(x) < eps
+
+ def _calc_currents_and_pf_from_pqu(
+ self,
+ voltages: list[float],
+ p_list: list[float],
+ q_list: list[float],
+ ) -> tuple[list[float], list[float]]:
+ """
+ Calculates signed currents (A) and signed power factors per phase from P/Q/U.
+
+ Convention:
+ - sign of current follows sign of active power P (import +, export -)
+ - PF = P / S (signed)
+ - S = sqrt(P^2 + Q^2)
+ - I = S / U (signed via P)
+ """
+ currents: list[float] = [0.0, 0.0, 0.0]
+ pfs: list[float] = [0.0, 0.0, 0.0]
+
+ for i in range(3):
+ u = voltages[i]
+ p = p_list[i]
+ q = q_list[i]
+
+ if self._nearly_zero(u):
+ currents[i] = 0.0
+ pfs[i] = 0.0
+ continue
+
+ s = math.sqrt(p * p + q * q)
+
+ if self._nearly_zero(s):
+ currents[i] = 0.0
+ pfs[i] = 0.0
+ continue
+
+ pfs[i] = p / s
+ i_mag = s / u
+ currents[i] = i_mag if p >= 0 else -i_mag
+
+ return currents, pfs
def update(self, client: PowerwallHttpClient, aggregate):
- # read firmware version
- status = client.get_json("/api/status")
- log.debug('Firmware: ' + status["version"])
try:
- # read additional info if firmware supports
meters_site = client.get_json("/api/meters/site")
+ cached = meters_site[0]["Cached_readings"]
+
+ voltages = [
+ float(cached[f"v_l{phase}n"])
+ for phase in range(1, 4)
+ ]
+
+ p_list = [
+ float(cached[f"real_power_{phase}"])
+ for phase in ["a", "b", "c"]
+ ]
+
+ q_list = [
+ float(cached[f"reactive_power_{phase}"])
+ for phase in ["a", "b", "c"]
+ ]
+
+ api_currents = [
+ float(cached[f"i_{phase}_current"])
+ for phase in ["a", "b", "c"]
+ ]
+
+ imported = aggregate["site"]["energy_imported"]
+ exported = aggregate["site"]["energy_exported"]
+
+ calculated_currents, power_factors = (
+ self._calc_currents_and_pf_from_pqu(
+ voltages=voltages,
+ p_list=p_list,
+ q_list=q_list,
+ )
+ )
+
+ if all(self._nearly_zero(i) for i in api_currents):
+ currents = calculated_currents
+ log.debug(
+ "Tesla/Neurio phase currents missing (all 0). "
+ "Calculated currents locally from P/Q and U."
+ )
+ else:
+ currents = api_currents
+ log.debug(
+ "Using phase currents from Tesla/Neurio API."
+ )
+
+ frequency = float(aggregate["site"]["frequency"])
+
+ serial = cached.get("serial_number")
+ serial_number = str(serial) if serial else None
+
powerwall_state = CounterState(
- imported=aggregate["site"]["energy_imported"],
- exported=aggregate["site"]["energy_exported"],
+ imported=imported,
+ exported=exported,
power=aggregate["site"]["instant_power"],
- voltages=[meters_site[0]["Cached_readings"]["v_l" + str(phase) + "n"] for phase in range(1, 4)],
- currents=[meters_site[0]["Cached_readings"]["i_" + phase + "_current"] for phase in ["a", "b", "c"]],
- powers=[meters_site[0]["Cached_readings"]["real_power_" + phase] for phase in ["a", "b", "c"]]
+ voltages=voltages,
+ currents=currents,
+ powers=p_list,
+ power_factors=power_factors,
+ frequency=round(frequency, 2),
+ serial_number=serial_number,
)
- except (KeyError, HTTPError):
+
+ except (KeyError, IndexError, TypeError, ValueError) as e:
log.debug(
- "Firmware seems not to provide detailed phase measurements. Fallback to total power only.")
+ "Firmware seems not to provide detailed phase measurements. "
+ "Fallback to total power only. (%s)",
+ str(e),
+ )
+
powerwall_state = CounterState(
imported=aggregate["site"]["energy_imported"],
exported=aggregate["site"]["energy_exported"],
- power=aggregate["site"]["instant_power"]
+ power=aggregate["site"]["instant_power"],
)
+
self.store.set(powerwall_state)
-component_descriptor = ComponentDescriptor(configuration_factory=TeslaCounterSetup)
+component_descriptor = ComponentDescriptor(
+ configuration_factory=TeslaCounterSetup
+)
diff --git a/packages/modules/devices/tesla/tesla/device.py b/packages/modules/devices/tesla/tesla/device.py
index 3a219cc1fe..590089c798 100644
--- a/packages/modules/devices/tesla/tesla/device.py
+++ b/packages/modules/devices/tesla/tesla/device.py
@@ -2,14 +2,23 @@
import logging
import requests
from requests import HTTPError
-from typing import Iterable, Union
+from typing import Iterable, Optional, Union
from modules.common.abstract_device import DeviceDescriptor
from modules.common.component_context import SingleComponentUpdateContext
-from modules.common.configurable_device import ComponentFactoryByType, ConfigurableDevice, MultiComponentUpdater
+from modules.common.configurable_device import (
+ ComponentFactoryByType,
+ ConfigurableDevice,
+ MultiComponentUpdater,
+)
from modules.common.req import get_http_session
from modules.devices.tesla.tesla.bat import TeslaBat
-from modules.devices.tesla.tesla.config import Tesla, TeslaBatSetup, TeslaCounterSetup, TeslaInverterSetup
+from modules.devices.tesla.tesla.config import (
+ Tesla,
+ TeslaBatSetup,
+ TeslaCounterSetup,
+ TeslaInverterSetup,
+)
from modules.devices.tesla.tesla.counter import TeslaCounter
from modules.devices.tesla.tesla.http_client import PowerwallHttpClient
from modules.devices.tesla.tesla.inverter import TeslaInverter
@@ -17,31 +26,47 @@
log = logging.getLogger(__name__)
-def __update_components(client: PowerwallHttpClient,
- components: Iterable[Union[TeslaBat, TeslaCounter, TeslaInverter]]):
+def __update_components(
+ client: PowerwallHttpClient,
+ components: Iterable[Union[TeslaBat, TeslaCounter, TeslaInverter]],
+):
aggregate = client.get_json("/api/meters/aggregates")
+
for component in components:
with SingleComponentUpdateContext(component.fault_state):
component.update(client, aggregate)
-def _authenticate(session: requests.Session, url: str, email: str, password: str):
+def _authenticate(
+ session: requests.Session,
+ url: str,
+ email: str,
+ password: str,
+):
"""
email is not yet required for login (2022/01), but we simulate the whole login page
"""
response = session.post(
"https://" + url + "/api/login/Basic",
- json={"username": "customer", "email": email, "password": password, "force_sm_off": False},
+ json={
+ "username": "customer",
+ "email": email,
+ "password": password,
+ "force_sm_off": False,
+ },
verify=False,
- timeout=5
+ timeout=5,
)
- log.debug("Authentication endpoint send cookies %s", str(response.cookies))
- return {"AuthCookie": response.cookies["AuthCookie"], "UserRecord": response.cookies["UserRecord"]}
+
+ return {
+ "AuthCookie": response.cookies["AuthCookie"],
+ "UserRecord": response.cookies["UserRecord"],
+ }
def create_device(device_config: Tesla):
- http_client = None
- session = None
+ http_client: Optional[PowerwallHttpClient] = None
+ session: Optional[requests.Session] = None
def create_bat_component(component_config: TeslaBatSetup):
return TeslaBat(component_config)
@@ -52,47 +77,103 @@ def create_counter_component(component_config: TeslaCounterSetup):
def create_inverter_component(component_config: TeslaInverterSetup):
return TeslaInverter(component_config)
- def update_components(components: Iterable[Union[TeslaBat, TeslaCounter, TeslaInverter]]):
+ def update_components(
+ components: Iterable[Union[TeslaBat, TeslaCounter, TeslaInverter]],
+ ):
log.debug("Beginning update")
+
address = device_config.configuration.ip_address
email = device_config.configuration.email
password = device_config.configuration.password
if http_client.cookies is None:
- http_client.cookies = _authenticate(session, address, email, password)
+ http_client.cookies = _authenticate(
+ session,
+ address,
+ email,
+ password,
+ )
__update_components(http_client, components)
return
+
try:
__update_components(http_client, components)
return
except HTTPError as e:
- if e.response.status_code != 401 and e.response.status_code != 403:
- raise e
- log.warning("Login to powerwall with existing cookie failed. Will retry with new cookie...")
- http_client.cookies = _authenticate(session, address, email, password)
+ status = getattr(
+ getattr(e, "response", None),
+ "status_code",
+ None,
+ )
+
+ if status not in (401, 403):
+ raise
+
+ log.warning(
+ "Login to powerwall with existing cookie failed "
+ "(status=%s). Will retry with new cookie...",
+ status,
+ )
+
+ http_client.cookies = _authenticate(
+ session,
+ address,
+ email,
+ password,
+ )
__update_components(http_client, components)
- log.debug("Update completed successfully")
def initializer():
nonlocal http_client, session
+
+ address = device_config.configuration.ip_address
+ email = device_config.configuration.email
+ password = device_config.configuration.password
+
session = get_http_session()
- http_client = PowerwallHttpClient(device_config.configuration.ip_address, session, None)
+
+ http_client = PowerwallHttpClient(
+ address,
+ session,
+ None,
+ )
+
+ http_client.cookies = _authenticate(
+ session,
+ address,
+ email,
+ password,
+ )
+
+ try:
+ status = http_client.get_json("/api/status")
+ log.debug("Firmware: %s", status["version"])
+ except (KeyError, requests.RequestException, ValueError):
+ log.debug(
+ "Could not read Tesla Powerwall firmware version.",
+ exc_info=True,
+ )
return ConfigurableDevice(
device_config=device_config,
initializer=initializer,
+ error_handler=initializer,
component_factory=ComponentFactoryByType(
bat=create_bat_component,
counter=create_counter_component,
inverter=create_inverter_component,
),
- component_updater=MultiComponentUpdater(update_components)
+ component_updater=MultiComponentUpdater(update_components),
)
device_descriptor = DeviceDescriptor(
configuration_factory=Tesla,
- compatibility_device_note="Tesla hat die lokale API Anfang 2025 mittels Firmwareupdate abgeschaltet.\nFür "
- "Neuanlagen muss aktuell auf unsere Zählerkits zurückgegriffen werden.",
- special_icon="ℹ️"
+ compatibility_device_note=(
+ "Tesla hat die lokale API mittels Firmwareupdate eingeschränkt.\n"
+ "Je nach Modell und Firmware-Version kann die Verwendung unserer "
+ "Zähler-Kits erforderlich sein, falls die direkte Anbindung nicht "
+ "funktioniert."
+ ),
+ special_icon="ℹ️",
)
diff --git a/packages/modules/devices/tesla/tesla/http_client.py b/packages/modules/devices/tesla/tesla/http_client.py
index f02eb15094..0d45803119 100644
--- a/packages/modules/devices/tesla/tesla/http_client.py
+++ b/packages/modules/devices/tesla/tesla/http_client.py
@@ -8,5 +8,10 @@ def __init__(self, host: str, session: requests.Session, cookies):
self.__session = session
def get_json(self, relative_url: str):
- url = self.__base_url + relative_url
- return self.__session.get(url, cookies=self.cookies, verify=False, timeout=5).json()
+ response = self.__session.get(
+ self.__base_url + relative_url,
+ cookies=self.cookies,
+ verify=False,
+ timeout=5,
+ )
+ return response.json()
diff --git a/packages/modules/devices/tesla/tesla/tesla.php b/packages/modules/devices/tesla/tesla/tesla.php
index ec396b638e..b428767d68 100644
--- a/packages/modules/devices/tesla/tesla/tesla.php
+++ b/packages/modules/devices/tesla/tesla/tesla.php
@@ -333,7 +333,7 @@ function teslaLogin () {
?>
Anmeldung erfolgreich!
- Die erhaltenen Token wurden gespeichert. Du kannst diese Seite jetzt schließen.
+ Die erhaltenen Token wurden gespeichert. Sie können diese Seite jetzt schließen.
- Gespeicherte Anmeldedaten wurden entfernt. Du kannst diese Seite jetzt schließen.
+ Gespeicherte Anmeldedaten wurden entfernt. Sie können diese Seite jetzt schließen.
-