-
Notifications
You must be signed in to change notification settings - Fork 27
/
exiConnector.py
362 lines (319 loc) · 15.2 KB
/
exiConnector.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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
# EXI is specified in w3.org/TR/exi
# For decoding and encoding, different decoders are availabe:
# 1. https://github.com/FlUxIuS/V2Gdecoder
# https://github.com/FlUxIuS/V2Gdecoder/releases
# Install java from https://www.java.com/de/download/manual.jsp
#
# C:\>java -version
# java version "1.8.0_351"
# Java(TM) SE Runtime Environment (build 1.8.0_351-b10)
# Java HotSpot(TM) Client VM (build 25.351-b10, mixed mode, sharing)
# C:\>
#
# java -jar V2Gdecoder.jar -e -s 8000dbab9371d3234b71d1b981899189d191818991d26b9b3a232b30020000040040
# ERROR: 'Premature EOS found while reading data.'
# <?xml version="1.0" encoding="UTF-8"?><ns4:supportedAppProtocolReq xmlns:ns4="urn:iso:15118:2:2010:AppProtocol"
# xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:ns3="http://www.w3.org/2001/XMLSchema">
# <AppProtocol>
# <ProtocolNamespace>urn:din:70121:2012:MsgDef</ProtocolNamespace>
# <VersionNumberMajor>2</VersionNumberMajor>
# <VersionNumberMinor>0</VersionNumberMinor>
# <SchemaID>1</SchemaID>
# <Priority>1</Priority>
# </AppProtocol>
# </ns4:supportedAppProtocolReq>
#
# 2. OpenV2G from https://github.com/Martin-P/OpenV2G
# Pro: The schema-specific files are already included as generated C-code, this
# makes it very fast.
# Contra: Is only a demo, does not provide an interface which could be used for decoding/encoding.
# Work in progress: Fork in https://github.com/uhi22/OpenV2Gx, to add a command line interface (and maybe a socket interface).
#
#
#
#
#
#
#
from helpers import showAsHex, twoCharHex
import subprocess
import sys
import time
import json
import os
# Example data:
# (1) From the Ioniq:
# In wireshark: Copy as hex stream
# 01fe8001000000228000dbab9371d3234b71d1b981899189d191818991d26b9b3a232b30020000040040
# remove the 8 bytes V2GTP header
# 8000dbab9371d3234b71d1b981899189d191818991d26b9b3a232b30020000040040
exiHexDemoSupportedApplicationProtocolRequestIoniq="8000dbab9371d3234b71d1b981899189d191818991d26b9b3a232b30020000040040"
# Command line:
# ./OpenV2G.exe DH8000dbab9371d3234b71d1b981899189d191818991d26b9b3a232b30020000040040
# (2) From OpenV2G main_example.appHandshake()
# 8000ebab9371d34b9b79d189a98989c1d191d191818981d26b9b3a232b30010000040001b75726e3a64696e3a37303132313a323031323a4d73674465660020000100880
exiHexDemoSupportedApplicationProtocolRequest2="8000ebab9371d34b9b79d189a98989c1d191d191818981d26b9b3a232b30010000040001b75726e3a64696e3a37303132313a323031323a4d73674465660020000100880"
# Command line:
# ./OpenV2G.exe DH8000ebab9371d34b9b79d189a98989c1d191d191818981d26b9b3a232b30010000040001b75726e3a64696e3a37303132313a323031323a4d73674465660020000100880
# (3) SupportedApplicationProtocolResponse
# 80400040
# Command line:
# ./OpenV2G.exe DH80400040
# (4) SessionSetupRequest DIN
# 809a0011d00000
# Command line:
# ./OpenV2G.exe DD809a0011d00000
# (5) SessionSetupResponse DIN
# 809a02004080c1014181c211e0000080
# ./OpenV2G.exe DD809a02004080c1014181c211e0000080
# (6) CableCheckReq
# "result": "809a001010400000"
# (7) PreChargeReq
# "result": "809a001150400000c80006400000"
# Configuration of the exi converter tool
if os.name == "nt":
# Windows: Path to the EXI decoder/encoder
pathToOpenV2GExe = "..\\OpenV2Gx\\Release\\OpenV2G.exe"
else:
# Linux: Path to the EXI decoder/encoder
pathToOpenV2GExe = "../OpenV2Gx/Release/OpenV2G.exe";
# Functions
def exiprint(s):
# todo: print to console or log or whatever
pass
def exiHexToByteArray(hexString):
# input: a string with the two-byte-hex representation
# output: a byte array with the same data.
# If the convertion fails, we return an empty byte array.
hexlen=len(hexString)
if ((hexlen % 2)!=0):
print("exiHexToByteArray: unplausible length of " + hexString)
return bytearray(0)
exiframe = bytearray(int(hexlen/2)) # two characters in the input string are one byte in the exiframe
for i in range(0, int(hexlen/2)):
x = hexString[2*i: 2*i+2]
#print("valuestring = " + x)
try:
v = int(x, 16)
#print("value " + hex(v))
exiframe[i] = v
except:
print("exiHexToByteArray: unplausible data " + x)
return bytearray(0)
return exiframe
def exiByteArrayToHex(b):
# input: Byte array
# output: a string with the two-byte-hex representation of the byte array
s = ""
for i in range(0, len(b)):
s = s + twoCharHex(b[i])
return s
def addV2GTPHeader(exidata):
#print("type is " + str(type(exidata)))
if (str(type(exidata)) == "<class 'str'>"):
#print("changing type to bytearray")
exidata = exiHexToByteArray(exidata)
#print("type is " + str(type(exidata)))
# takes the bytearray with exidata, and adds a header to it, according to the Vehicle-to-Grid-Transport-Protocol
exiLen = len(exidata)
header = bytearray(8) # V2GTP header has 8 bytes
# 1 byte protocol version
# 1 byte protocol version inverted
# 2 bytes payload type
# 4 byte payload length
header[0] = 0x01 # version
header[1] = 0xfe # version inverted
header[2] = 0x80 # payload type. 0x8001 means "EXI data"
header[3] = 0x01 #
header[4] = (exiLen >> 24) & 0xff # length 4 byte.
header[5] = (exiLen >> 16) & 0xff
header[6] = (exiLen >> 8) & 0xff
header[7] = exiLen & 0xff
return header + exidata
def removeV2GTPHeader(v2gtpData):
#removeV2GTPHeader
return v2gtpData[8:]
def exiDecode(exiHex, prefix="DH"):
# input: exi data. Either hexstring, or bytearray or bytes
# prefix to select the schema
# if the input is a byte array, we convert it into hex string. If it is already a hex string, we take it as it is.
#print("type is " + str(type(exiHex)))
if (str(type(exiHex)) == "<class 'bytearray'>"):
#print("changing type to hex string")
exiHex = exiByteArrayToHex(exiHex)
if (str(type(exiHex)) == "<class 'bytes'>"):
#print("changing type to hex string")
exiHex = exiByteArrayToHex(exiHex)
#print("type is " + str(type(exiHex)))
param1 = prefix + exiHex # DH for decode handshake
#print("exiDecode: trying to decode " + exiHex + " with schema " + prefix)
result = subprocess.run(
[pathToOpenV2GExe, param1], capture_output=True, text=True)
#print("stdout:", result.stdout)
if (len(result.stderr)>0):
print("exiDecode ERROR. stderr:" + result.stderr)
strConverterResult = result.stdout
return strConverterResult
def exiEncode(strMessageName):
# todo: handle the schema, the message name and the parameters
# param1 = "Eh" # Eh for encode handshake, SupportedApplicationProtocolResponse
# param1 = "EDa" # EDa for Encode, Din, SessionSetupResponse
param1 = strMessageName
exiprint("[EXICONNECTOR] exiEncode " + param1)
result = subprocess.run([pathToOpenV2GExe, param1], capture_output=True, text=True)
if (len(result.stderr)>0):
strConverterResult = "exiEncode ERROR. stderr:" + result.stderr
print(strConverterResult)
else:
#print("exiEncode stdout:", result.stdout)
# Now we have an encoder result in json form, something like:
# {
# "info": "",
# "error": "",
# "result": "8004440400"
# }
try:
y = json.loads(result.stdout)
strConverterResult = y["result"]
strConverterError = y["error"]
if (len(strConverterError)>0):
print("[EXICONNECTOR] exiEncode error " + strConverterError)
#print("strConverterResult is " + str(strConverterResult))
except:
strConverterResult = "exiEncode failed to convert json to dict."
print(strConverterResult)
return strConverterResult
def testByteArrayConversion(s):
print("Testing conversion of " + s)
x = exiHexToByteArray(s)
newHexString = exiByteArrayToHex(x)
print("exi as hex=" + newHexString)
exiWithHeader = addV2GTPHeader(x)
exiWithHeaderString = exiByteArrayToHex(exiWithHeader)
print("with V2GTP header=" + exiWithHeaderString)
def testDecoder(strHex, pre="DH", comment=""):
global nFail
strHex = strHex.replace(" ", "") # remove blanks
print("Decoder test for " + comment + " with data " + strHex)
decoded=exiDecode(strHex, pre)
print(decoded)
if (len(comment)>0):
strExpected = comment
if (decoded.find(strExpected)>0):
print("---pass---")
else:
print("---***!!!FAIL!!!***---")
nFail+=1
def testReadExiFromSnifferFile():
file1 = open('results\\tmp.txt', 'r')
Lines = file1.readlines()
for myLine in Lines:
if (myLine[0:9]=="[SNIFFER]"):
posOfEqualsign = myLine.find("=")
s = myLine[posOfEqualsign+1:] # The part after the "=" contains the EXI hex data.
s = s.replace(" ", "") # Remove blanks
s = s.replace("\n", "") # Remove line feeds
#print(s)
testDecoder(s, "DD", "")
def testReadExiFromExiLogFile():
file1 = open('PevExiLog.txt', 'r')
fileOut = open('PevExiLog.decoded.txt', 'w')
# example: "ED 809a02004080c1014181c210b8"
# example with timestamp: "2022-12-20T08:17:15.055755=ED 809a02004080c1014181c21198"
Lines = file1.readlines()
for myLine in Lines:
posOfEqual=myLine.find("=")
if (posOfEqual>0):
# we have an equal-sign. Take the string behind it.
strToDecode=myLine[posOfEqual+1:]
else:
# no equal-sign. Take the complete line.
strToDecode=myLine
if (strToDecode[1:3]=="D "): # it is DIN
posOfSpace=2
s = strToDecode[posOfSpace+1:] # The part after the " " contains the EXI hex data.
s = s.replace(" ", "") # Remove blanks
s = s.replace("\n", "") # Remove line feeds
#print(s)
decoded=exiDecode(s, "DD")
print(decoded)
print(myLine.replace("\n", "") + " means:", file=fileOut)
print(decoded, file=fileOut)
fileOut.close()
def testTimeConsumption():
strHex = "809a001150400000c80006400000"
pre = "DD"
tStart = time.time()
nRuns = 100
for i in range(0, nRuns):
decoded=exiDecode(strHex, pre)
tStop = time.time()
elapsed_time = tStop - tStart
print("Decoder: Execution time for " + str(nRuns) + " runs:", elapsed_time, "seconds")
tStart = time.time()
nRuns = 100
for i in range(0, nRuns):
s = exiEncode("EDC_1122334455667788")
tStop = time.time()
elapsed_time = tStop - tStart
print("Encoder: Execution time for " + str(nRuns) + " runs:", elapsed_time, "seconds")
if __name__ == "__main__":
nFail=0
print("Testing exiConnector...")
if (False):
testTimeConsumption()
exit()
if (False):
testReadExiFromExiLogFile()
exit()
if (False):
testDecoder("8000ebab9371d34b9b79d189a98989c1d191d191818981d26b9b3a232b30010000040001b75726e3a64696e3a37303132313a323031323a4d73674465660020000100880", pre="DH", comment="supportedAppProtocolReq")
testDecoder("80400040", pre="DH", comment="supportedAppProtocolRes")
testDecoder("809a0011d00000", pre="DD", comment="SessionSetupReq")
testDecoder("809a02004080c1014181c211e0000080", pre="DD", comment="SessionSetupRes")
testDecoder("809a001198", pre="DD", comment="ServiceDiscoveryReq")
testDecoder("809a0011a0012002412104", pre="DD", comment="ServiceDiscoveryRes")
testDecoder("809a0011b2001280", pre="DD", comment="ServicePaymentSelectionReq")
testDecoder("809a0011c000", pre="DD", comment="ServicePaymentSelectionRes")
testDecoder("809a00107211400dc0c8c82324701900", pre="DD", comment="ChargeParameterDiscoveryReq")
testDecoder("809a001080004820400000c99002062050193080c0c802064c8010190140c80a20", pre="DD", comment="ChargeParameterDiscoveryRes")
testDecoder("809a001010400000", pre="DD", comment="CableCheckReq")
testDecoder("809a0010200200000000", pre="DD", comment="CableCheckRes")
testDecoder("809a001150400000c80006400000", pre="DD", comment="PreChargeReq")
testDecoder("809a00116002000000320000", pre="DD", comment="PreChargeRes")
if (False):
print("The request from the Ioniq after the EVSE sent ServicePaymentSelectionRes:")
testDecoder("809A00113020000A00000000", pre="DD", comment="PowerDeliveryReq")
if (True):
print("The session setup request of the Ioniq:")
testDecoder("809A02000000000000000011D01811959401930C00", pre="DD", comment="SessionSetupReq")
if (False):
testDecoder("80 9A 02 00 40 80 C1 01 41 81 C2 11 E0 00 00 80", pre="DD", comment="SessionSetupRes")
testDecoder("80 9A 02 00 40 80 C1 01 41 81 C2 11 94 00", pre="DD", comment="ServiceDiscoveryReq")
testDecoder("80 9A 02 00 40 80 C1 01 41 81 C2 11 A0 01 20 02 41 00 84", pre="DD", comment="ServiceDiscoveryRes")
testDecoder("80 9A 02 00 40 80 C1 01 41 81 C2 11 B2 00 12 80", pre="DD", comment="ServicePaymentSelectionReq")
testDecoder("80 9A 02 00 40 80 C1 01 41 81 C2 11 C0 00", pre="DD", comment="ServicePaymentSelectionRes")
print("The error response of the Ioniq with FAILED_ChargingSystemIncompatibility")
testDecoder("80 9A 02 00 40 80 C1 01 41 81 C2 11 30 20 00 0A 00 00 00 00", pre="DD", comment="PowerDeliveryReq")
print("The request of the Ioniq after ServicePaymentSelectionResponse")
testDecoder("80 9A 02 00 40 80 C1 01 41 81 C2 10 B8", pre="DD", comment="ContractAuthenticationReq")
if (False):
print("The response of the Alpi chargeParameterDiscovery")
testDecoder("80 9A 02 00 00 00 00 03 1F DC 8B D0 80 02 00 00 00 00 00 10 00 2A 80 04 00 00 14 0C 00 40 80 E1 8A 3A 0A 0A 00 A0 60 60 00 08 0A 01 E2 30 30 06 10", pre="DD", comment="ChargeParameterDiscoveryRes")
if (False):
testDecoder("809a00113060", pre="DD", comment="PowerDeliveryReq")
testDecoder("809a0011400420400000", pre="DD", comment="PowerDeliveryRes")
testDecoder("809a0010d0400000c800410c8000", pre="DD", comment="CurrentDemandReq")
testDecoder("809a0010e0020000003200019000000600", pre="DD", comment="CurrentDemandRes")
testDecoder("809a001210400000", pre="DD", comment="WeldingDetectionReq")
testDecoder("809a00122002000000320000", pre="DD", comment="WeldingDetectionRes")
testDecoder("809a0011f0", pre="DD", comment="SessionStopReq")
testDecoder("809a00120000", pre="DD", comment="SessionStopRes")
print("Number of fails: " + str(nFail))
#print("Testing exiDecode with exiHexDemoSupportedApplicationProtocolRequestIoniq")
#print(exiDecode(exiHexDemoSupportedApplicationProtocolRequestIoniq))
#print("Testing exiDecode with exiHexDemoSupportedApplicationProtocolRequest2")
#strConverterResult = exiDecode(exiHexDemoSupportedApplicationProtocolRequest2)
#print(strConverterResult)
#strConverterResult = exiDecode(exiHexToByteArray(exiHexDemoSupportedApplicationProtocolRequest2))
#print(strConverterResult)