github-mirrors/pyPLC
1
0
Fork 0
mirror of https://github.com/uhi22/pyPLC.git synced 2024-09-11 00:38:29 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

extend soc_callback hook with extra info

Browse source
This commit is contained in:
Jorrit Pouw 2023-06-26 19:38:45 +02:00
parent b22c82faf4
commit df274ee085
3 changed files with 51 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
doc/pyPlc.ini.template
View file

@ -106,3 +106,5 @@ charge_parameter_backend = chademo
# REST callback for SoC states. Comment out to disable. Do not leave a trailing slash
soc_callback_enabled = False
soc_callback_endpoint = http://1.1.1.1
# Fallback value to use if the vehicle does not support the EVEnergyCapacity.Value
soc_fallback_energy_capacity = 2700

34
evseNoGui.py
View file

@ -21,14 +21,38 @@ def cbShowStatus(s, selection=""):
soc_callback_enabled = getConfigValueBool("soc_callback_enabled")
soc_callback_url = getConfigValue("soc_callback_endpoint") if soc_callback_enabled else ""
soc_fallback_energy_capacity = getConfigValue("soc_fallback_energy_capacity")
def socStatusCallback(current_soc: int, full_soc: int = -1, energy_capacity: int = -1, energy_request: int = -1, evccid: str = "", origin: str = ""):
# Do some basic value checks and conversions
# Some cars do not support certain values and return 0, make sure we actually send -1
if (energy_capacity > 0):
# We need Wh, not something in between kWh and Wh
energy_capacity = energy_capacity * 10
else:
# Some cars do not supply energy capacity of their battery
# Support some kind of fallback value which would work for installations where typically one car charges
if (int(soc_fallback_energy_capacity) > 0):
energy_capacity = int(soc_fallback_energy_capacity) * 10
else:
energy_capacity = -1
if (energy_request > 0):
# We need Wh, not something in between kWh and Wh
energy_request = energy_request * 10
else:
energy_request = -1
def socStatusCallback(remaining_soc: int, full_soc: int = -1, bulk_soc: int = -1, origin: str = ""):
print(f"Received SoC status from {origin}.\n"
f" Remaining {remaining_soc}% \n"
f" Full at {full_soc}%\n"
f" Bulk at {bulk_soc}%")
f" Current SoC {current_soc}% \n"
f" Full SoC {full_soc}%\n"
f" Energy capacity {energy_capacity} Wh \n"
f" Energy request {energy_request} Wh \n"
f" EVCCID {evccid} \n")
if soc_callback_enabled:
requests.post(f"{soc_callback_url}/modem?remaining_soc={remaining_soc}&full_soc={full_soc}&bulk_soc={bulk_soc}")
requests.post(f"{soc_callback_url}?current_soc={current_soc}&full_soc={full_soc}&energy_capacity={energy_capacity}&energy_request={energy_request}&evccid={evccid}")
myMode = C_EVSE_MODE

34
fsmEvse.py
View file

@ -28,9 +28,9 @@ class fsmEvse():
def publishStatus(self, s):
self.callbackShowStatus(s, "evseState")
def publishSoCs(self, remaining_soc: int, full_soc: int = -1, bulk_soc: int = -1, origin: str = ""):
def publishSoCs(self, current_soc: int, full_soc: int = -1, energy_capacity: int = -1, energy_request: int = -1, evccid: str = "", origin: str = ""):
if self.callbackSoCStatus is not None:
self.callbackSoCStatus(remaining_soc, full_soc, bulk_soc, origin)
self.callbackSoCStatus(current_soc, full_soc, energy_capacity, energy_request, self.evccid, origin)
def enterState(self, n):
self.addToTrace("from " + str(self.state) + " entering " + str(n))
@ -82,6 +82,9 @@ class fsmEvse():
self.Tcp.transmit(msg)
self.publishStatus("Session established")
self.enterState(stateWaitForServiceDiscoveryRequest)
y = json.loads(strConverterResult)
self.evccid = y.get("EVCCID", "")
if (self.isTooLong()):
self.enterState(0)
@ -136,8 +139,8 @@ class fsmEvse():
# todo: check the request content, and fill response parameters
self.addToTrace("Received PowerDeliveryReq. Extracting SoC parameters")
info = json.loads(strConverterResult)
remaining_soc = int(info.get("EVRESSSOC", -1))
self.publishSoCs(remaining_soc, origin="PowerDeliveryReq")
current_soc = int(info.get("EVRESSSOC", -1))
self.publishSoCs(current_soc, origin="PowerDeliveryReq")
msg = addV2GTPHeader(exiEncode("EDh")) # EDh for Encode, Din, PowerDeliveryResponse
if (testsuite_faultinjection_is_triggered(TC_EVSE_ResponseCode_Failed_for_PowerDeliveryRes)):
# send a PowerDeliveryResponse with Responsecode Failed
@ -149,10 +152,11 @@ class fsmEvse():
if (strConverterResult.find("ChargeParameterDiscoveryReq")>0):
self.addToTrace("Received ChargeParameterDiscoveryReq. Extracting SoC parameters via DC")
info = json.loads(strConverterResult)
remaining_soc = int(info.get("DC_EVStatus.EVRESSSOC", -1))
current_soc = int(info.get("DC_EVStatus.EVRESSSOC", -1))
full_soc = int(info.get("FullSOC", -1))
bulk_soc = int(info.get("BulkSOC", -1))
self.publishSoCs(remaining_soc, full_soc, bulk_soc, origin="ChargeParameterDiscoveryReq")
energy_capacity = int(info.get("EVEnergyCapacity.Value", -1))
energy_request = int(info.get("EVEnergyRequest.Value", -1))
self.publishSoCs(current_soc, full_soc, energy_capacity, energy_request, origin="ChargeParameterDiscoveryReq")
# todo: check the request content, and fill response parameters
msg = addV2GTPHeader(exiEncode("EDe")) # EDe for Encode, Din, ChargeParameterDiscoveryResponse
@ -168,8 +172,8 @@ class fsmEvse():
# todo: make a real cable check, and while it is ongoing, send "Ongoing".
self.addToTrace("Received CableCheckReq. Extracting SoC parameters via DC")
info = json.loads(strConverterResult)
remaining_soc = int(info.get("DC_EVStatus.EVRESSSOC", -1))
self.publishSoCs(remaining_soc, -1, -1, origin="CableCheckReq")
current_soc = int(info.get("DC_EVStatus.EVRESSSOC", -1))
self.publishSoCs(current_soc, -1, -1, origin="CableCheckReq")
msg = addV2GTPHeader(exiEncode("EDf")) # EDf for Encode, Din, CableCheckResponse
if (testsuite_faultinjection_is_triggered(TC_EVSE_ResponseCode_Failed_for_CableCheckRes)):
@ -232,13 +236,14 @@ class fsmEvse():
strEVTargetVoltageMultiplier = y["EVTargetVoltage.Multiplier"]
uTarget = combineValueAndMultiplier(strEVTargetVoltageValue, strEVTargetVoltageMultiplier)
self.addToTrace("EV wants EVTargetVoltage " + str(uTarget))
remaining_soc = int(y.get("DC_EVStatus.EVRESSSOC", -1))
current_soc = int(y.get("DC_EVStatus.EVRESSSOC", -1))
full_soc = int(y.get("FullSOC", -1))
bulk_soc = int(y.get("BulkSOC", -1))
self.publishSoCs(remaining_soc, full_soc, bulk_soc, origin="CurrentDemandReq")
energy_capacity = int(y.get("EVEnergyCapacity.Value", -1))
energy_request = int(y.get("EVEnergyRequest.Value", -1))
self.callbackShowStatus(str(remaining_soc), "soc")
self.publishSoCs(current_soc, full_soc, energy_capacity, energy_request, origin="CurrentDemandReq")
self.callbackShowStatus(str(current_soc), "soc")
except:
self.addToTrace("ERROR: Could not decode the CurrentDemandReq")
@ -352,6 +357,7 @@ class fsmEvse():
self.state = 0
self.cyclesInState = 0
self.rxData = []
self.evccid = ""
def mainfunction(self):
self.Tcp.mainfunction() # call the lower-level worker