-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathchargepoints.py
139 lines (129 loc) · 7.59 KB
/
chargepoints.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
"""
chargepoints.py
Stellt die für Ladepunkte notwendigen Daten und Methoden zur Verfügung.
part of project soc_helper
Copyright (C) 2023-2024 M. Williges (spam at zut punkt de)
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from dataclasses import dataclass
import logging
import json
import datetime
import cars
import spritmonitor
import configuration
import energylog
@dataclass
class chargepoint:
# Variablen
chargepointId: int = 1 # Default-Nummer des Ladepunktes
plugstate: bool = False # Zustand Ladestecker
counterAtPlugin: float = None # Zählerstand beim Stecken des Stecker
counter: float = None # zuletzt übermittelter Zählerstand
connectedId: int = None # aktuell im Ladepunkt eingestelltes Fahrzeug
# Hilfsfunktionen
def getCounterTopic(self):
return('openWB/chargepoint/'+str(self.chargepointId)+'/get/imported')
def getPlugStateTopic(self):
return('openWB/chargepoint/'+str(self.chargepointId)+'/get/plug_state')
def getConnectedIdTopic(self):
return('openWB/chargepoint/'+str(self.chargepointId)+'/get/connected_vehicle/info')
# Callback-Methoden
def cb_energycounter(self, client, userdata, msg):
# Gesamtzählerstand
self.counter = float(msg.payload)
logging.debug(f'Zählerstand auf Ladepunkt {self.chargepointId}: {self.counter}')
def cb_connectedVehicle(self, client, userdata, msg):
# ID des eingestellten Fahrzeugs
# Überdenken: Diverse Daten werden bei Einbuchen ins WLAN übertragen, bevor man die Möglichkeit hat,
# das Fahrzeug an der Wallbox ggf richtig einzustellen. Daher sollte connectedId nur ausgewertet werden,
# wenn der Stecker gezogen wird.
self.connectedId = json.loads(msg.payload)['id']
logging.debug(f'ID des mit Ladepunkt {self.chargepointId} verbundenes Fahrzeugs: {self.connectedId}')
def cb_plug(self, client, userdata, msg):
# Stecker des Ladepunktes
logging.debug(f'Steckerzustand von Ladepunkt {self.chargepointId}: {msg.payload}')
if (b'true' in msg.payload) or (b'1' in msg.payload):
if not self.plugstate:
# Wechsel von 0 auf 1 (Anstecken)
self.plugstate = True
self.counterAtPlugin = self.counter
# In der Wallbox gewähltes Fahrzeug heraussuchen und dessen SoC als SoC zu Ladebeginn merken
found = False
for car in configuration.myCars:
if car.openwbVehicleId == self.connectedId:
logging.debug(f'Angestecktes Fahrzeug mit openwbId {self.connectedId} als {car.name} in der Fahrzeugliste identifiziert')
found = True
car.socAtPlugin = car.soc
logging.info(f'Angestecktes Fahrzeug {car.name}, ID {self.connectedId} mit SoC {car.socAtPlugin} angesteckt')
break
if not found:
logging.warn(f'in der Wallbox eingestelltes Fahrzeug mit ID {self.connectedId} ist nicht in der Fahrzeugliste aufgeführt')
else:
if self.plugstate:
# Wechsel von 1 auf 0 (Abstecken)
self.plugstate = False
# Auf kWh umrechnen und mit 3 Nachkommastellen runden
if self.counter is not None and self.counterAtPlugin is not None:
lastCharged = round( (self.counter - self.counterAtPlugin)/1000, 3)
else:
logging.warn(f'Aktueller Stromzählerstand oder Zählerstand beim Anstecken war undefiniert.')
lastCharged = 0;
logging.info(f'Stecker von Ladepunkt {self.chargepointId} gezogen. Lademenge {lastCharged} geladen in Fahrzeug {self.connectedId}.')
# Fahrzeugindex herausfinden
found = False
for car in configuration.myCars:
if car.openwbVehicleId == self.connectedId:
logging.debug(f'Abgestecktes Fahrzeug mit openwbId {self.connectedId} als {car.name} in der Fahrzeugliste identifiziert')
found = True
break
if not found:
logging.warn(f'in der Wallbox eingestelltes Fahrzeug mit ID {self.connectedId} ist nicht in der Fahrzeugliste aufgeführt')
# Zeitpunkt des Endes des Ladevorganges
datetimeAtPlugOut = datetime.datetime.now() # Datum des Ladeendes
date = datetimeAtPlugOut.strftime('%d.%m.%Y') # Datum in lesbarer Form für lokale Datei und ggf. Spritmonitor
if found:
# Fahrzeug in Liste gefunden
# lokales Abspeichern
energylog.write(date+', '+car.name+', '+str(car.odo)+', '+str(lastCharged)+', '+str(car.socAtPlugin)+', '+str(car.openwbsoc)+'\n')
# hochladen zu Spritmonitor
if car.useSpritmonitor and (lastCharged >= 0.1):
logging.debug(f'Spritmonitor ist konfiguriert. Beginne Übermittlung.')
# letzten bei Spritmonitor eingetragenen km-Stand auslesen
lastFueling = spritmonitor.get_last_fuel_entry(car.spritmonitorVehicleId)
if len(lastFueling) > 0:
td = json.loads(json.dumps(lastFueling[0]))
lastOdo = float(td['odometer'])
logging.debug(f'Letzter Spritmonitor-Kilometerstand ist {lastOdo}.')
else:
lastOdo = 0.0
logging.warning(f'Konnte keinen letzter Spritmonitor-Kilometerstand ermitteln, nehme 0 an.')
logging.info(f'Ladung wurde mit berechneten {car.openwbsoc}% beendet.')
# type-Variable für Füllung bestimmen
if lastOdo == 0.0:
fuel_type = "first"
elif ( abs(float(car.openwbsoc)-100)<=2.0 ):
fuel_type = "full"
else:
fuel_type = "notfull"
trip = round(car.odo - lastOdo,2)
if trip < 0.1:
trip = 0.0
logging.info(f'Letzter Spritmonitor-Kilometerstand: {lastOdo}; aktueller Kilometerstand: {car.odo} trip: {trip}')
quantityunitid = 5 # 'kWh'
# "Betankung" übermitteln
result=spritmonitor.add_fuel_entry(car.spritmonitorVehicleId, 1,date, fuel_type, car.odo, trip, lastCharged,
quantityunitid, car.spritmonitorFuelsort, car.spritmonitorFuelprice,car.openwbsoc,car.spritmonitorAttributes)
logging.debug(f'Ergebnis der Übermittlung: {result}')
logging.info(f'Übermittlung an Spritmonitor ist erfolgt.')
else:
# undefiniertes Fahrzeug angesteckt
energylog.write(date+', undefiniertes Fahrzeug, -,'+str(lastCharged)+', -, -\n')