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

fixed address handling on raspberry

Browse source
This commit is contained in:
uhi22 2022-11-20 18:47:55 +01:00
parent f61f68a0ab
commit c792162015
3 changed files with 97 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
addressManager.py
View file

@ -35,6 +35,7 @@ class addressManager():
# If we have multiple interfaces (e.g. ethernet and WLAN), it will find multiple link-local-addresses.
foundAddresses = []
if os.name == 'nt':
# on Windows
result = subprocess.run(["ipconfig.exe"], capture_output=True, text=True, encoding="ansi")
if (len(result.stderr)>0):
print(result.stderr)
@ -45,6 +46,23 @@ class addressManager():
k = line.find(" fe80::")
if (k>0):
foundAddresses.append(line[k+1:])
else:
# on Raspberry
result = subprocess.run(["ifconfig"], capture_output=True, text=True)
if (len(result.stderr)>0):
print(result.stderr)
else:
lines = result.stdout.split("\n")
for line in lines:
if (line.find("inet6")>0):
k = line.find(" fe80::") # the beginning of the IPv6
if (k>0):
sIpWithText = line[k+1:]
x = sIpWithText.find(" ") # the space is the end of the IPv6
sIp = sIpWithText[0:x]
# print("[addressManager] IP=>" + sIp + "<")
foundAddresses.append(sIp)
print("[addressManager] Found " + str(len(foundAddresses)) + " link-local IPv6 addresses.")
for a in foundAddresses:
print(a)
@ -63,9 +81,39 @@ class addressManager():
def findLocalMacAddress(self):
# Find out the MAC address of the local ethernet interface.
# Todo: Find this out dynamically.
self.localMac = MAC_LAPTOP
print("[addressManager] we have local MAC " + prettyMac(self.localMac) + ". Todo: find this out dynamically.")
ba = bytearray(6)
if os.name == 'nt':
# on Windows
self.localMac = MAC_LAPTOP
print("[addressManager] we have local MAC " + prettyMac(self.localMac) + ". Todo: find this out dynamically.")
else:
# on raspberry
# We use "ifconfig", and search for the line which says "ether <mac> ..."
result = subprocess.run(["ifconfig"], capture_output=True, text=True)
if (len(result.stderr)>0):
print(result.stderr)
else:
lines = result.stdout.split("\n")
for line in lines:
if (line.find(" ether ")>0) and (line.find("(Ethernet)")>0):
# print(line)
k = line.find(" ether ")
# print(k)
strMac = line[k+7:k+24]
# e.g. "b8:27:eb:12:34:56"
# print(strMac)
# Remove all ":"
strMac = strMac.replace(":", "")
#print(strMac)
if (len(strMac)!=12):
print("[addressManager] ERROR: invalid length of MAC string. Expected be 6 bytes, means 12 hex characters. Found " + str(len(strMac)))
else:
for i in range(0, 6):
sTwoChar = strMac[2*i : 2*i+2]
ba[i] = int(sTwoChar, 16)
self.localMac = ba
print("[addressManager] we have local MAC " + prettyMac(self.localMac) + ".")
def setPevMac(self, pevMac):
# During the SLAC, the MAC of the PEV was found out. Store it, maybe we need it later.
self.pevMac = pevMac
@ -166,6 +214,6 @@ if __name__ == "__main__":
am = addressManager()
#am.setPevIp(bytearray([0xfe, 0x80]))
am.setPevIp(bytearray([0xfe, 0x80, 0x00, 0x00, 0xfe, 0x80, 0x00, 0x00, 0xfe, 0x80, 0x00, 0x00, 0xDE, 0xAD, 0xBE, 0xEF]))
print(am.getLinkLocalIpv6Address())
print(am.getLinkLocalIpv6Address(resulttype="bytearray"))
print("Test result: LinkLocalIpv6=" + am.getLinkLocalIpv6Address())
print("same as byte array: " + str(am.getLinkLocalIpv6Address(resulttype="bytearray")))

5
pyPlcHomeplug.py
View file

@ -865,9 +865,7 @@ class pyPlcHomeplug():
self.sniffer.setnonblock(True)
self.NMK = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16 ] # a default network key
self.NID = [ 1, 2, 3, 4, 5, 6, 7 ] # a default network ID
# self.pevMac = [0x55, 0x56, 0x57, 0x58, 0x59, 0x5A ] # a default pev MAC. Will be overwritten later.
#self.pevMac = [0x04, 0x65, 0x65, 0x00, 0x64, 0xC3 ] # todo: in case of PevMode, use the MAC from the OS.
self.pevMac = [0xDC, 0x0E, 0xA1, 0x11, 0x67, 0x08 ] # todo: in case of PevMode, use the MAC from the OS.
self.pevMac = [0xDC, 0x0E, 0xA1, 0x11, 0x67, 0x08 ] # a default pev MAC. Will be overwritten later.
self.evseMac = [0x55, 0x56, 0x57, 0xAA, 0xAA, 0xAA ] # a default evse MAC. Will be overwritten later.
self.myMAC = self.addressManager.getLocalMacAddress()
self.runningCounter=0
@ -879,6 +877,7 @@ class pyPlcHomeplug():
self.enterEvseMode()
if (mode == C_PEV_MODE):
self.enterPevMode()
self.pevMac = self.myMAC
self.showStatus(prettyMac(self.pevMac), "pevmac")
print("sniffer created at " + self.strInterfaceName)

55
pyPlcTcpSocket.py
View file

@ -28,8 +28,34 @@ class pyPlcTcpClientSocket():
# for connecting, we are still in blocking-mode because
# otherwise we run into error "[Errno 10035] A non-blocking socket operation could not be completed immediately"
# We set a shorter timeout, so we do not block too long if the connection is not established:
print("step1")
self.sock.settimeout(0.5)
self.sock.connect((host, port))
print("step2")
# https://stackoverflow.com/questions/71022092/python3-socket-program-udp-ipv6-bind-function-with-link-local-ip-address-gi
# While on Windows10 just connecting to a remote link-local-address works, under
# Raspbian the socket.connect says "invalid argument".
# host = "2003:d1:170f:d500:1744:89d:d921:b20f" # works
# host = "2003:d1:170f:d500:2052:326e:cef9:ad07" # works
# host = "fe80::c690:83f3:fbcb:980e" # invalid argument. Link local address needs an interface specified.
# host = "fe80::c690:83f3:fbcb:980e%eth0" # ok with socket.getaddrinfo
if (os.name != 'nt'):
# We are at the Raspberry
print(host[0:5].lower())
if (host[0:5].lower()=="fe80:"):
print("This is a link local address. We need to add %eth0 at the end.")
host = host + "%eth0"
socket_addr = socket.getaddrinfo(host,port,socket.AF_INET6,socket.SOCK_DGRAM,socket.SOL_UDP)[0][4]
print(socket_addr)
print("step2b")
# https://stackoverflow.com/questions/5358021/establishing-an-ipv6-connection-using-sockets-in-python
# (address, port, flow info, scope id) 4-tuple for AF_INET6
# On Raspberry, the ScopeId is important. Just giving 0 leads to "invalid argument" in case
# of link-local ip-address.
#self.sock.connect((host, port, 0, 0))
self.sock.connect(socket_addr)
print("step3")
self.sock.setblocking(0) # make this socket non-blocking, so that the recv function will immediately return
self.isConnected = True
except socket.error as e:
@ -204,20 +230,23 @@ def testServerSocket():
def testClientSocket():
print("Testing the pyPlcTcpClientSocket...")
c = pyPlcTcpClientSocket()
c.connect('fe80::e0ad:99ac:52eb:85d3', 15118)
#c.connect('fe80::e0ad:99ac:52eb:85d3', 15118)
#c.connect('fe80::e0ad:99ac:52eb:9999', 15118)
#c.connect('localhost', 15118)
c.connect('fe80::c690:83f3:fbcb:980e', 15118)
print("connected="+str(c.isConnected))
print("sending something to the server")
c.transmit(bytes("Test", "utf-8"))
for i in range(0, 10):
print("waiting 1s")
time.sleep(1)
if (c.isRxDataAvailable()):
d = c.getRxData()
print("received " + str(d))
if ((i % 3)==0):
print("sending something to the server")
c.transmit(bytes("Test", "utf-8"))
if (c.isConnected):
print("sending something to the server")
c.transmit(bytes("Test", "utf-8"))
for i in range(0, 10):
print("waiting 1s")
time.sleep(1)
if (c.isRxDataAvailable()):
d = c.getRxData()
print("received " + str(d))
if ((i % 3)==0):
print("sending something to the server")
c.transmit(bytes("Test", "utf-8"))
print("end")