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

added modes for the ipv6 module. correct dest MAC in SDR response.

Browse source
This commit is contained in:
uhi22 2022-10-21 09:02:37 +02:00
parent 9fa7890f88
commit 16c30bd3b4
2 changed files with 39 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
pyPlcHomeplug.py
View file

@ -542,14 +542,17 @@ class pyPlcHomeplug():
def enterPevMode(self): def enterPevMode(self):
self.iAmEvse = 0 # not emulating a charging station self.iAmEvse = 0 # not emulating a charging station
self.iAmPev = 1 # emulating a vehicle self.iAmPev = 1 # emulating a vehicle
self.ipv6.enterPevMode()
self.showStatus("PEV mode", "mode") self.showStatus("PEV mode", "mode")
def enterEvseMode(self): def enterEvseMode(self):
self.iAmEvse = 1 # not emulating a charging station self.iAmEvse = 1 # emulating a charging station
self.iAmPev = 0 # emulating a vehicle self.iAmPev = 0 # not emulating a vehicle
self.ipv6.enterEvseMode()
self.showStatus("EVSE mode", "mode") self.showStatus("EVSE mode", "mode")
def enterListenMode(self): def enterListenMode(self):
self.iAmEvse = 0 # not emulating a charging station self.iAmEvse = 0 # not emulating a charging station
self.iAmPev = 0 # emulating a vehicle self.iAmPev = 0 # not emulating a vehicle
self.ipv6.enterListenMode()
self.showStatus("LISTEN mode", "mode") self.showStatus("LISTEN mode", "mode")
def __init__(self, callbackAddToTrace=None, callbackShowStatus=None): def __init__(self, callbackAddToTrace=None, callbackShowStatus=None):

54
pyPlcIpv6.py
View file

@ -3,6 +3,11 @@
# #
# It has the following sub-functionalities: # It has the following sub-functionalities:
# - IP.UDP.SDP: listen to requests from the car, and responding to them. # - IP.UDP.SDP: listen to requests from the car, and responding to them.
# Eth --> IPv6 --> UDP --> V2GTP --> SDP
# |
# v
# Eth <-- IPv6 <-- UDP <-- V2GTP <-- SDP
#
# #
# Abbreviations: # Abbreviations:
# SECC: Supply Equipment Communication Controller. The "computer" of the charging station. # SECC: Supply Equipment Communication Controller. The "computer" of the charging station.
@ -22,15 +27,11 @@ class ipv6handler():
def packResponseIntoEthernet(self, buffer): def packResponseIntoEthernet(self, buffer):
# packs the IP packet into an ethernet packet # packs the IP packet into an ethernet packet
self.EthResponse = bytearray(len(buffer) + 6 + 6 + 2) # Ethernet header needs 14 bytes: self.EthResponse = bytearray(len(buffer) + 6 + 6 + 2) # Ethernet header needs 14 bytes:
# 6 bytes MAC # 6 bytes destination MAC
# 6 bytes MAC # 6 bytes source MAC
# 2 bytes EtherType # 2 bytes EtherType
self.EthResponse[0] = 0x33 # destination MAC. We use multicast. Todo: Better use the PEV MAC. for i in range(0, 6): # fill the destination MAC with the source MAC of the received package
self.EthResponse[1] = 0x33 self.EthResponse[i] = self.myreceivebuffer[6+i]
self.EthResponse[2] = 0x00
self.EthResponse[3] = 0x00
self.EthResponse[4] = 0x00
self.EthResponse[5] = 0x01
self.fillMac(self.ownMac) # bytes 6 to 11 are the source MAC self.fillMac(self.ownMac) # bytes 6 to 11 are the source MAC
self.EthResponse[12] = 0x86 # 86dd is IPv6 self.EthResponse[12] = 0x86 # 86dd is IPv6
self.EthResponse[13] = 0xdd self.EthResponse[13] = 0xdd
@ -79,7 +80,7 @@ class ipv6handler():
lenInclChecksum = len(buffer) + 8 lenInclChecksum = len(buffer) + 8
self.UdpResponse[4] = lenInclChecksum >> 8 self.UdpResponse[4] = lenInclChecksum >> 8
self.UdpResponse[5] = lenInclChecksum & 0xFF self.UdpResponse[5] = lenInclChecksum & 0xFF
checksum = 0x1234 # todo: calculate this checksum checksum = 0x1234 # todo: calculate this checksum, see https://en.wikipedia.org/wiki/User_Datagram_Protocol
self.UdpResponse[6] = checksum >> 8 self.UdpResponse[6] = checksum >> 8
self.UdpResponse[7] = checksum & 0xFF self.UdpResponse[7] = checksum & 0xFF
for i in range(0, len(buffer)): for i in range(0, len(buffer)):
@ -87,10 +88,10 @@ class ipv6handler():
showAsHex(self.UdpResponse, "UDP response ") showAsHex(self.UdpResponse, "UDP response ")
self.packResponseIntoIp(self.UdpResponse) self.packResponseIntoIp(self.UdpResponse)
def prepareSdpResponse(self): def sendSdpResponse(self):
# SECC Discovery Response. # SECC Discovery Response.
# The response from the charger to the EV, which tells the chargers IPv6 address to the EV. # The response from the charger to the EV, which transfers the IPv6 address of the charger to the car.
self.SdpPayload = bytearray(20) self.SdpPayload = bytearray(20) # SDP response has 20 bytes
for i in range(0, 16): for i in range(0, 16):
self.SdpPayload[i] = self.SeccIp[i] # 16 bytes IP address of the charger self.SdpPayload[i] = self.SeccIp[i] # 16 bytes IP address of the charger
self.SdpPayload[16] = 15118 >> 8 # SECC port high byte. Port is always 15118. self.SdpPayload[16] = 15118 >> 8 # SECC port high byte. Port is always 15118.
@ -118,10 +119,11 @@ class ipv6handler():
showAsHex(self.V2Gframe, "V2Gframe ") showAsHex(self.V2Gframe, "V2Gframe ")
self.packResponseIntoUdp(self.V2Gframe) self.packResponseIntoUdp(self.V2Gframe)
def evaluateUdp(self): def evaluateUdpPayload(self):
if (self.destinationport == 15118): # port for the SECC if (self.destinationport == 15118): # port for the SECC
if ((self.udpPayload[0]==0x01) and (self.udpPayload[1]==0xFE)): # protocol version 1 and inverted if ((self.udpPayload[0]==0x01) and (self.udpPayload[1]==0xFE)): # protocol version 1 and inverted
# it is a V2GTP message # it is a V2GTP message
showAsHex(self.udpPayload, "V2GTP ")
self.evccPort = self.sourceport self.evccPort = self.sourceport
v2gptPayloadType = self.udpPayload[2] * 256 + self.udpPayload[3] v2gptPayloadType = self.udpPayload[2] * 256 + self.udpPayload[3]
# 0x8001 EXI encoded V2G message # 0x8001 EXI encoded V2G message
@ -139,16 +141,25 @@ class ipv6handler():
if (seccDiscoveryReqTransportProtocol!=0x00): if (seccDiscoveryReqTransportProtocol!=0x00):
print("seccDiscoveryReqTransportProtocol " + str(seccDiscoveryReqTransportProtocol) + " is not supported") print("seccDiscoveryReqTransportProtocol " + str(seccDiscoveryReqTransportProtocol) + " is not supported")
else: else:
# This was a valid SDP request. Let's respond. # This was a valid SDP request. Let's respond, if we are the charger.
print("ok, this was a valid SDP request")
showAsHex(self.udpPayload, "udp payload ") if (self.iAmEvse==1):
print("ok, this was a valid SDP request. Will respond.") print("We are the SECC. Sending SDP response.")
self.prepareSdpResponse() self.sendSdpResponse()
else: else:
print("v2gptPayloadLen on SDP request is " + str(v2gptPayloadLen) + " not supported") print("v2gptPayloadLen on SDP request is " + str(v2gptPayloadLen) + " not supported")
else: else:
print("v2gptPayloadType " + hex(v2gptPayloadType) + " not supported") print("v2gptPayloadType " + hex(v2gptPayloadType) + " not supported")
def enterPevMode(self):
self.iAmEvse = 0 # not emulating a charging station
self.iAmPev = 1 # emulating a vehicle
def enterEvseMode(self):
self.iAmEvse = 1 # emulating a charging station
self.iAmPev = 0 # not emulating a vehicle
def enterListenMode(self):
self.iAmEvse = 0 # not emulating a charging station
self.iAmPev = 0 # not emulating a vehicle
def evaluateReceivedPacket(self, pkt): def evaluateReceivedPacket(self, pkt):
if (len(pkt)>60): if (len(pkt)>60):
@ -162,14 +173,15 @@ class ipv6handler():
# udplen is including 8 bytes header at the begin # udplen is including 8 bytes header at the begin
if (self.udplen>8): if (self.udplen>8):
self.udpPayload = bytearray(self.udplen-8) self.udpPayload = bytearray(self.udplen-8)
print("self.udplen=" + str(self.udplen)) # print("self.udplen=" + str(self.udplen))
print("self.myreceivebuffer len=" + str(len(self.myreceivebuffer))) # print("self.myreceivebuffer len=" + str(len(self.myreceivebuffer)))
for i in range(0, self.udplen-8): for i in range(0, self.udplen-8):
#print("index " + str(i) + " " + hex(self.myreceivebuffer[62+i])) #print("index " + str(i) + " " + hex(self.myreceivebuffer[62+i]))
self.udpPayload[i] = self.myreceivebuffer[62+i] self.udpPayload[i] = self.myreceivebuffer[62+i]
self.evaluateUdp() self.evaluateUdpPayload()
def __init__(self, transmitCallback): def __init__(self, transmitCallback):
self.enterEvseMode()
self.transmit = transmitCallback self.transmit = transmitCallback
self.SeccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0x06, 0xaa, 0xaa, 0xff, 0xfe, 0, 0xaa, 0xaa ] # 16 bytes, a default IPv6 address for the charging station self.SeccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0x06, 0xaa, 0xaa, 0xff, 0xfe, 0, 0xaa, 0xaa ] # 16 bytes, a default IPv6 address for the charging station
self.EvccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0x06, 0x65, 0x65, 0xff, 0xfe, 0, 0x64, 0xC3 ] # 16 bytes, a default IPv6 address for the vehicle self.EvccIp = [ 0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0x06, 0x65, 0x65, 0xff, 0xfe, 0, 0x64, 0xC3 ] # 16 bytes, a default IPv6 address for the vehicle