Remove delay from CAN transmit routine

.gitignore

@@ -1,2 +1,5 @@
 __pycache__/
 *.py[ocd]
+
+env/
+.venv/

Test.py

@@ -1,17 +1,21 @@
 import time
 import can
 
-bustype = 'socketcan'
+bustype = "socketcan"
-channel = 'can0'  #echten can nutzen
+channel = "can0"  # echten can nutzen
+
 
 def producer(id):
     """:param id: Spam the bus with messages including the data id."""
     bus = can.Bus(channel=channel, interface=bustype)
     for i in range(10):
-        msg = can.Message(arbitration_id=0x002, data=[id, i, 0, 1, 3, 1, 4, 1], is_extended_id=False)
+        msg = can.Message(
+            arbitration_id=0x002, data=[id, i, 0, 1, 3, 1, 4, 1], is_extended_id=False
+        )
         bus.send(msg)
 
     time.sleep(1)
-   
+
-producer(10) # rausfinden welche id der Slave hat
+
+producer(10)  # rausfinden welche id der Slave hat
 print("I am done with this shit")

TestRead.py

@@ -3,13 +3,12 @@ import can
 
 input("startReading")
 
-can_interface = 'can0' #echten can nutzen
+can_interface = "can0"  # echten can nutzen
-bus = can.interface.Bus(can_interface, bustype='socketcan')
+bus = can.interface.Bus(can_interface, bustype="socketcan")
 message = bus.recv(5.0)
 
 if message is None:
-    print('Timeout occurred, no message.')
+    print("Timeout occurred, no message.")
-else :
+else:
-	print('We got a message hell yeah')
+    print("We got a message hell yeah")
-	print("{}".format(message.data)) #hat gesendete data von can benutzen
+    print("{}".format(message.data))  # hat gesendete data von can benutzen
-

canTest.py

@@ -5,89 +5,98 @@ import can
 waitForUserInput = True
 bus = None
 
-def sendMessageOnCan(message, bustype='socketcan',channel='can0'):
-	
-	global bus
-	if bus is None:
-		bus = can.interface.Bus(channel, bustype='socketcan')
-		busEnabled = True
 
-	msg = can.Message(arbitration_id=0x002, data=message, is_extended_id=False)
+def sendMessageOnCan(message, bustype="socketcan", channel="can0"):
-	bus.send(msg)
+    global bus
-	time.sleep(1)
+    if bus is None:
+        bus = can.interface.Bus(channel, bustype="socketcan")
+        busEnabled = True
 
-def recive(bustype='socketcan',channel='can0'):
+    msg = can.Message(arbitration_id=0x002, data=message, is_extended_id=False)
-	global bus	
+    bus.send(msg)
-	message = bus.recv()
-
-	if message is None:
-		print('Timeout occurred, no message.')
-		#timeOut
-
-	else :
-		#print('We got a message hell yeah')
-		print(message)
-		return message
-
-def waitForUserInput(waitForUserInput):
-	if(waitForUserInput):
-		readInput = "n"
-		print("Ready to start ? ")
-		while('y' not in readInput):
-			print("Enter y to start or c to cancel ")
-			readInput = input(" ")
-			if(readInput == 'c'):
-				return -1
-	return 0
-
-def CanTest(bustype='socketcan',channel='can0'):
-#-----------------------------------------------------------------------------------------------------#
-	print("Start testing the can bus connectivity ")
-	global waitForUserInput
-	if (waitForUserInput(waitForUserInput) != 0):
-		return
-
-	print("testing can for a single message")
-	messageType = 1
-	testMessage = [messageType, 0xc, 0, 0xf, 0xf, 0xe, 0xe]
-	testMessageData = [1, 12, 0, 15, 15, 14, 14, 0]
-	
-	sendMessageOnCan(testMessage)
-	recivedMessage = recive().data
-
-	result = True
-	count = 0
-
-	for entry in recivedMessage :
-		if entry != testMessageData[count]:
-			result = False
-			break
-
-		count = count + 1
-
-	print(recivedMessage)
-	print("Sending a single message test with answer result {}".format(result))
-#-----------------------------------------------------------------------------------------------------#
-	
-	print("testing for multiple sends")
-	if (waitForUserInput(waitForUserInput) != 0):
-		return
-
-	packete = []
-	sendCountFinally = 0
-	trials = 12
-	for sendCount in range(trials):
-		sendMessageOnCan([1, 0xc, 0, 0xf, 0xf, 0xe, 0xe, sendCount])
-		time.sleep(0.1)
-		packete.append(recive())
-
-	for packet in packete:	
-		print(packet.data)
-
-	print("Sending multiple testmessages test with answer result {}".format(len(packete) == len(range(trials))))
-	
-#-----------------------------------------------------------------------------------------------------#
-	print("I am done with this shit")
 
 
+def recive(bustype="socketcan", channel="can0"):
+    global bus
+    message = bus.recv()
 
+    if message is None:
+        print("Timeout occurred, no message.")
+        # timeOut
+
+    else:
+        # print('We got a message hell yeah')
+        print(message)
+        return message
+
+
+def waitForUserInput(waitForUserInput, prompt="Ready to start?"):
+    if waitForUserInput:
+        readInput = "n"
+        print(prompt)
+        while "y" not in readInput:
+            print("Enter y to start or c to cancel ")
+            readInput = input(" ")
+            if readInput == "c":
+                return -1
+    return 0
+
+
+def CanTest(bustype="socketcan", channel="can0"):
+    # -----------------------------------------------------------------------------------------------------#
+    print("====================")
+    print("Start testing the can bus connectivity ")
+    global waitForUserInput
+    if waitForUserInput(waitForUserInput) != 0:
+        return
+
+    print("testing can for a single message")
+    messageType = 1
+    testMessage = [messageType, 0xC, 0, 0xF, 0xF, 0xE, 0xE]
+    testMessageData = [1, 12, 0, 15, 15, 14, 14, 0]
+
+    sendMessageOnCan(testMessage)
+    recivedMessage = recive().data
+
+    result = True
+    count = 0
+
+    for entry in recivedMessage:
+        if entry != testMessageData[count]:
+            result = False
+            break
+
+        count = count + 1
+
+    print(recivedMessage)
+    print(
+        "Sending a single message test with result {}".format(
+            "PASSED" if result else "FAILED"
+        )
+    )
+    # -----------------------------------------------------------------------------------------------------#
+
+    print("====================")
+    print("testing for multiple sends")
+    if waitForUserInput(waitForUserInput) != 0:
+        return
+
+    packete = []
+    sendCountFinally = 0
+    trials = 12
+    for sendCount in range(trials):
+        sendMessageOnCan([1, 0xC, 0, 0xF, 0xF, 0xE, 0xE, sendCount])
+        time.sleep(0.1)
+        packete.append(recive())
+
+    for packet in packete:
+        print(packet.data)
+
+    print(
+        "Sending multiple testmessages test with result {}".format(
+            "PASSED" if len(packete) == len(range(trials)) else "FAILED"
+        )
+    )
+
+    # -----------------------------------------------------------------------------------------------------#
+    print("I am done with this shit")

eProm.py

@@ -1,12 +1,14 @@
 import canTest
 
-def ePromTest():
-	print("start eprom test")
-	message = [4, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,0x00]
-	canTest.waitForUserInput(True)
-	canTest.sendMessageOnCan(message)
-	answer = canTest.recive();
-	print("ePromTest reult {}".format(str(answer.data) == "bytearray(b'iiiiBiii')"))
-	
 
-	print("I am done with this shit")
+def ePromTest():
+    print("====================")
+    print("start eprom test")
+    message = [4, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]
+    canTest.waitForUserInput(True)
+    canTest.sendMessageOnCan(message)
+    answer = canTest.recive()
+    result = answer.data == bytearray(b"iiiBiiii")
+    print("ePromTest result {}".format("PASSED" if result else "FAILED"))
+
+    print("I am done with this shit")

realTest.py

@@ -5,18 +5,19 @@ import eProm
 
 
 def startBalancingTest():
-	message = [5, 0xc, 0, 0xf, 0xf, 0xe, 0xe]
+    message = [5, 0xC, 0, 0xF, 0xF, 0xE, 0xE]
-	print("startBalancing")
+    print("====================")
-	canTest.waitForUserInput(True)
+    print("startBalancing")
-	canTest.sendMessageOnCan(message)
+    canTest.waitForUserInput(True)
+    canTest.sendMessageOnCan(message)
 
 
-#canTest.CanTest()
+canTest.CanTest()
-#eProm.ePromTest()
+eProm.ePromTest()
-#voltageTest.voltagesTest()
+voltageTest.voltagesTest()
-tempTest.tempTest() 
+tempTest.tempTest()
 
-#startBalancingTest()
+startBalancingTest()
 """
 print("start Temp test")
 message = [3, 0xc, 0, 0xf, 0xf, 0xe, 0xe]
@@ -33,5 +34,3 @@ canTest.sendMessageOnCan(message)
 answer = canTest.recive();
 print(answer.data)
 """
-
-

requirements.txt

@@ -0,0 +1,10 @@
+black==23.3.0
+click==8.1.3
+msgpack==1.0.5
+mypy-extensions==1.0.0
+packaging==23.0
+pathspec==0.11.1
+platformdirs==3.2.0
+python-can==4.1.0
+typing_extensions==4.5.0
+wrapt==1.15.0

tempTest.py

@@ -3,15 +3,18 @@ import can
 import canTest
 import struct
 
-bustype = 'socketcan'
+bustype = "socketcan"
-channel = 'can0'
+channel = "can0"
+
 
 def getTempOverCan():
     bus = can.Bus(channel=channel, interface=bustype)
-    message = can.Message(arbitration_id=0x002, data=[3, 0, 0, 0, 0, 0, 0], is_extended_id=False)
+    message = can.Message(
+        arbitration_id=0x002, data=[3, 0, 0, 0, 0, 0, 0], is_extended_id=False
+    )
     bus.send(message)
     voltages = []
-    runtime  =  2
+    runtime = 2
     begin = 0
 
     for msg in bus:
@@ -19,42 +22,46 @@ def getTempOverCan():
         voltages.append(msg.data)
         print("begin {}".format(begin))
         begin = begin + 1
-        if(begin > runtime):
+        if begin > runtime:
             return voltages
 
+
 def verifyNumbers(numberListList):
     zeroCount = 0
     max = 0
     min = 1000000
-    for numberList in numberListList: 
+    for numberList in numberListList:
-    	print("Temperatures {}".format(numberList))
+        print("Temperatures {}".format(numberList))
-    	for number in numberList:
+        for number in numberList:
-
             if number < min:
                 min = number
 
-            elif number > max :   
+            elif number > max:
                 max = number
 
     print("Temperatures {}".format())
     print("The biggest difference tmep was max{} and min{}".format((max), (min)))
 
-def tempTest():  
-    print("starting temperature test ")
-    canTest.waitForUserInput(True)
-    Temps = getTempOverCan() 
-    allTemp = []
-    changed = []
 
-    for temp in Temps:
+def tempTest():
-        allTemp.append(struct.unpack("<HHHH", temp)) #Format anpassen
+    while True:
+        print("====================")
+        print("starting temperature test ")
+        canTest.waitForUserInput(True)
+        Temps = getTempOverCan()
+        allTemp = []
+        changed = []
 
-    print("all Number")    
+        for temp in Temps:
-    print(allTemp)    
+            allTemp += struct.unpack("<HHHH", temp)  # Format anpassen
 
-    for te in allTemp:    
+        print("Raw temperatures:")
-        for con in te:
+        print(allTemp)
-            changed.append(((con/16)*0.0625))
+
-    print(changed)        
+        for te in allTemp:
+            changed.append(((te >> 4) * 0.0625))
+        print("Interpreted temperatures:")
+        print(changed)
+        if canTest.waitForUserInput(True, "Repeat temperature test?") != 0:
+            break
     print("I am done with this shit")
-

voltageTest.py

@@ -3,51 +3,63 @@ import can
 import canTest
 import struct
 
-bustype = 'socketcan'
+bustype = "socketcan"
-channel = 'can0'
+channel = "can0"
+
 
 def getBatteryVoltageOverCan():
     bus = can.Bus(channel=channel, interface=bustype)
-    message = can.Message(arbitration_id=0x002, data=[2, 0, 0, 0, 0, 0, 0], is_extended_id=False)
+    message = can.Message(
+        arbitration_id=0x002, data=[2, 0, 0, 0, 0, 0, 0], is_extended_id=False
+    )
     bus.send(message)
     voltages = []
-    runtime  =  3
+    runtime = 3
     begin = 0
 
     for msg in bus:
-        #print(msg.data)
+        # print(msg.data)
         voltages.append(msg.data)
-        #print("begin {}".format(begin))
+        # print("begin {}".format(begin))
         begin = begin + 1
-        if(begin > 4):
+        if begin > 4:
             return voltages
 
+
 def verifyNumbers(numberListList):
     zeroCount = 0
     max = 0
     min = 1000000
-    for numberList in numberListList: 
+    for numberList in numberListList:
         for number in numberList:
             if number == 0:
-                zeroCount +=1
+                zeroCount += 1
 
             elif number < min:
                 min = number
 
-            elif number > max :   
+            elif number > max:
                 max = number
 
     print("There were 3 expected Zeros and {} detected".format(zeroCount))
-    print("The biggest difference were over {} and under {}".format((max-35000), (35000-min)))
+    print(
+        "The biggest difference were over {} and under {}".format(
+            (max - 35000), (35000 - min)
+        )
+    )
 
-def voltagesTest():  
+
+def voltagesTest():
+    print("====================")
     print("starting voltage test pls but 3.500 as reference value")
-    canTest.waitForUserInput(True)
+    while True:
-    voltages = getBatteryVoltageOverCan() # rausfinden welche id der Slave hat
+        canTest.waitForUserInput(True)
-    allVoltages = []
+        voltages = getBatteryVoltageOverCan()  # rausfinden welche id der Slave hat
-    for volt in voltages:
+        allVoltages = []
-        allVoltages.append(struct.unpack("<HHHH", volt))
+        for volt in voltages:
-    
+            allVoltages.append(struct.unpack("<HHHH", volt))
-    verifyNumbers(allVoltages)        
-    print("I am done with this shit")
 
+        verifyNumbers(allVoltages)
+        print("I am done with this shit")
+        if canTest.waitForUserInput(True, "Repeat voltage test?") != 0:
+            break