Ich habe anhand des von dir verlinkten Codes die Änderungen eingearbeitet und bei mir getestet... es hat mich einfach genervt, dass ich kein echtes phasengenaues Lastmanagement hatte!
Im Details sind folgende Änderungen eingearbeitet:
- Wenn alle Phasenströme 0 sind und P/Q/U vorliegen, die Ströme berechnen
- Leistungsfaktor wird berechnet: cosφ = P / S
Das Ergebnis in Verbindung einer Tesla Powerwall mit einer OpenWB:
- Plausible Phasenströme (Abweichung zu Tesla One ~0,03–0,05 A)
- Lastmanagement & Schieflast funktionieren korrekt
- Keine Änderung für Setups mit echten Stromwerten (greift nur bei I = 0)
Spricht aus eurer Sicht etwas dagegen, diese Logik in den Trunk aufzunehmen? Und könnte dies ggf. jemand übernehmen. Ich möchte mich nicht in eurer Github einarbeiten...
Anbei der überarbeitete Code der counter.py:
Code: Alles auswählen
#!/usr/bin/env python3
import logging
import math
from requests import HTTPError
from modules.common.abstract_device import AbstractCounter
from modules.common.component_state import CounterState
from modules.common.component_type import ComponentDescriptor
from modules.common.fault_state import ComponentInfo, FaultState
from modules.common.store import get_counter_value_store
from modules.devices.tesla.tesla.config import TeslaCounterSetup
from modules.devices.tesla.tesla.http_client import PowerwallHttpClient
log = logging.getLogger(__name__)
class TeslaCounter(AbstractCounter):
def __init__(self, component_config: TeslaCounterSetup) -> None:
self.component_config = component_config
def initialize(self) -> None:
self.store = get_counter_value_store(self.component_config.id)
self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))
def update(self, client: PowerwallHttpClient, aggregate):
# read firmware version
status = client.get_json("/api/status")
log.debug('Firmware: ' + status.get("version", "unknown"))
try:
# read additional info if firmware supports
meters_site = client.get_json("/api/meters/site")
cached = meters_site[0]["Cached_readings"]
# per-phase voltages and active powers
voltages = [float(cached["v_l" + str(phase) + "n"]) for phase in range(1, 4)]
powers = [float(cached["real_power_" + phase]) for phase in ["a", "b", "c"]]
# reactive power may be missing in some firmwares
reactive = []
for phase in ["a", "b", "c"]:
q = cached.get("reactive_power_" + phase)
try:
reactive.append(float(q) if q is not None else None)
except (TypeError, ValueError):
reactive.append(None)
# currents from API (often 0 for Neurio in some Tesla firmwares)
currents = []
for phase in ["a", "b", "c"]:
i = cached.get("i_" + phase + "_current", 0)
try:
currents.append(float(i) if i is not None else 0.0)
except (TypeError, ValueError):
currents.append(0.0)
# --- calculate power factors per phase (PF = P / S) ---
# S = sqrt(P^2 + Q^2); if Q missing -> assume Q=0 => S=abs(P)
power_factors = []
for p, q in zip(powers, reactive):
s = math.hypot(p, q) if q is not None else abs(p)
pf = (p / s) if s > 1e-6 else 0.0
# If you want PF always positive (0..1), use:
# pf = abs(pf)
power_factors.append(pf)
# --- if currents are missing (all 0), calculate them from P/Q/U ---
if all(abs(i) < 1e-3 for i in currents):
calc_currents = []
for p, u, q in zip(powers, voltages, reactive):
if u <= 1e-6:
calc_currents.append(0.0)
continue
s = math.hypot(p, q) if q is not None else abs(p)
i = (s / u) if s > 0 else 0.0
# Keep direction consistent with P (import +, export -)
if p < 0:
i = -abs(i)
else:
i = abs(i)
calc_currents.append(i)
if any(abs(i) > 1e-3 for i in calc_currents):
currents = calc_currents
log.debug(
"Tesla/Neurio phase currents missing (all 0). Calculated currents from P/Q and U."
)
powerwall_state = CounterState(
imported=aggregate["site"]["energy_imported"],
exported=aggregate["site"]["energy_exported"],
power=aggregate["site"]["instant_power"],
voltages=voltages,
currents=currents,
powers=powers,
power_factors=power_factors,
)
except (KeyError, HTTPError, IndexError, TypeError, ValueError):
log.debug("Firmware seems not to provide detailed phase measurements. Fallback to total power only.")
powerwall_state = CounterState(
imported=aggregate["site"]["energy_imported"],
exported=aggregate["site"]["energy_exported"],
power=aggregate["site"]["instant_power"],
)
self.store.set(powerwall_state)
component_descriptor = ComponentDescriptor(configuration_factory=TeslaCounterSetup)
Anbei noch ein Screenshot des Ergebnis:
- Unbenannt.png (18.5 KiB) 23 mal betrachtet
Vielen Dank und viele Grüße
NicoP.