pyPLC/claralogConverter.py

# claralogConverter
#
# This little helper tool reads a text log file containing V2G messages (e.g. from https://github.com/uhi22/ccs32clara) and
# interprets the content of the EXI data.
#
# Preconditions:
# 1. You have a log file which contains V2G traffic (e.g. from https://openinverter.org/forum/viewtopic.php?p=64655#p64655)
# 2. You cloned and compiled the OpenV2Gx EXI decoder from https://github.com/uhi22/OpenV2Gx 
#
# Limitations:
# - Only DIN is supported at the moment.
# - The script treats all V2G EXI messages as DIN messages. This means, the ApplHandshake messages at
#   the begin of the charging session will lead to wrong or not decoded data.
# - The path where the script look for pcap files needs to be configured in the code.
#
# Possible improvements / Todos:
# - Show also the SLAC, NeigborDiscovery and SDP.
# - Add flexibility to also decode the ApplHandshake messages.
# - Add ISO support.
# - Configure the path where to look for pcap files via command line
#

import exiConnector
import os
from helpers import combineValueAndMultiplier
import json

# The path where the script will search for pcap files:
directory = 'local/pcaps_to_convert'

# stop the evaluation after this number of packets. Set to zero to have no limit.
nLimitNumberOfPackets = -1


# Examples of the decoder result:
#"EVSEPresentVoltage.Multiplier": "0",
#"EVSEPresentVoltage.Value": "318",
#"EVSEPresentVoltage.Unit": "V",
#"DC_EVStatus.EVRESSSOC": "53",

def convertClaralogToTxt(inputFileName):
    global nLimitNumberOfPackets
    global directory
    fileIn = open(inputFileName, 'r')
    fileOut = open(inputFileName + '.decoded.txt', 'w')
    print("# generated by claralogConverter.py", file=fileOut)
    print("# https://github.com/uhi22/pyPLC", file=fileOut)
    fileOutValues = open(inputFileName + '.values.txt', 'w')
    print("# generated by claralogConverter.py", file=fileOutValues)
    print("# https://github.com/uhi22/pyPLC", file=fileOutValues)
    fileOutStatistics = open(directory + '/pcap_statistics.txt', 'a')
    print("# statistics for " + inputFileName, file=fileOutStatistics)
    t1CableCheckBegin = 0
    t2PreChargeBegin = 0
    t3CurrentDemandBegin = 0
    numberOfPackets=0
    Lines = fileIn.readlines()
    for line in Lines:
        numberOfPackets+=1
        #print(packet)
        print(line.strip(), file=fileOut)
        posTcpPayload = line.find(": 01 fe 80 01 00")
        if posTcpPayload>0:
                # we found the V2GTP header.
                tcppayload = line[posTcpPayload + 2:] # everything until the end of the line is the data
                # this gives a string of hex values, separated by " ", e.g. "01 fe 80 01"
                s = tcppayload.replace(" ", "") # remove spaces
                if (s[0:8]=="01fe8001"):
                    # it is a V2GTP header with EXI content
                    strExi = s[16:] # remove V2GTP header (8 bytes, means 16 hex characters)
                    pre = "DD" # decode DIN
                    decoded=exiConnector.exiDecode(strExi, pre)
                    #print(decoded)
                    print(decoded, file=fileOut)
                    print(decoded)
                    strTimeStamp = line[line.find("[")+1 : line.find("]")]
                    print(strTimeStamp)
                    try:
                        jsondict = json.loads(decoded)
                        try:
                            u = combineValueAndMultiplier(jsondict["EVSEPresentVoltage.Value"], jsondict["EVSEPresentVoltage.Multiplier"])
                            print("[" + strTimeStamp + "] EVSEPresentVoltage=" + str(u), file=fileOutValues)
                            i = combineValueAndMultiplier(jsondict["EVSEPresentCurrent.Value"], jsondict["EVSEPresentCurrent.Multiplier"])
                            print("[" + strTimeStamp + "] EVSEPresentCurrent=" + str(i), file=fileOutValues)
                        except:
                            print("no present voltage")
                            pass
                        try:
                            u = combineValueAndMultiplier(jsondict["EVTargetVoltage.Value"], jsondict["EVTargetVoltage.Multiplier"])
                            print("[" + strTimeStamp + "] EVTargetVoltage=" + str(u), file=fileOutValues)
                            i = combineValueAndMultiplier(jsondict["EVTargetCurrent.Value"], jsondict["EVTargetCurrent.Multiplier"])
                            print("[" + strTimeStamp + "] EVTargetCurrent=" + str(i), file=fileOutValues)
                        except:
                            pass
                        try:
                            soc = jsondict["DC_EVStatus.EVRESSSOC"]
                            print("[" + strTimeStamp + "] EVRESSSOC=" + str(soc), file=fileOutValues)
                        except:
                            pass
                    except:
                        pass
        if ((numberOfPackets % 100)==0):
            print(str(numberOfPackets) + " packets")
        if ((nLimitNumberOfPackets>0) and (numberOfPackets>=nLimitNumberOfPackets)):
            break
    # Statistics of the timing:
    #print("t1CableCheckBegin " + str(t1CableCheckBegin))
    #print("t2PreChargeBegin " + str(t2PreChargeBegin))
    #print("t3CurrentDemandBegin " + str(t3CurrentDemandBegin))
    if ((t1CableCheckBegin>0) and (t2PreChargeBegin>t1CableCheckBegin) and (t3CurrentDemandBegin>t2PreChargeBegin)):
        print("charger MAC " + chargerMAC + " " + getManufacturerFromMAC(chargerMAC))
        timeForCableCheck = t2PreChargeBegin - t1CableCheckBegin
        timeForPreCharge = t3CurrentDemandBegin - t2PreChargeBegin
        print("timeForCableCheck= " + ("%.3f" % timeForCableCheck))
        print("timeForPreCharge= " + ("%.3f" % timeForPreCharge))

        print(chargerMAC + ";" + getManufacturerFromMAC(chargerMAC) + ";" + \
        "timeForCableCheck;" + ("%.3f" % timeForCableCheck) + ";" + \
        "timeForPreCharge; " + ("%.3f" % timeForPreCharge), file=fileOutStatistics)
        
    fileOutStatistics.close()
    fileOut.close()
    fileOutValues.close()
    fileIn.close()


# iterate over files in the directory
for filename in os.listdir(directory):
    strFileNameWithPath = os.path.join(directory, filename)
    # checking if it is a file
    if os.path.isfile(strFileNameWithPath):
        print(strFileNameWithPath)
        # check the file extension:
        if (strFileNameWithPath[-9:]==".claralog"):
            print("Will decode " + strFileNameWithPath)
            convertClaralogToTxt(strFileNameWithPath)