add jungle support to bulk test
#!/usr/bin/env python3
-import threading
+import os
import time
+import threading
from typing import Any, List
from panda import Panda
+JUNGLE = "JUNGLE" in os.environ
+if JUNGLE:
+ from panda_jungle import PandaJungle
+
# The TX buffers on pandas is 0x100 in length.
NUM_MESSAGES_PER_BUS = 10000
@@ -17,14 +22,18 @@ def flood_tx(panda):
if __name__ == "__main__":
serials = Panda.list()
+ if JUNGLE:
+ sender = Panda()
+ receiver = PandaJungle()
+ else:
if len(serials) != 2:
raise Exception("Connect two pandas to perform this test!")
-
sender = Panda(serials[0])
receiver = Panda(serials[1])
+ receiver.set_safety_mode(Panda.SAFETY_ALLOUTPUT)
sender.set_safety_mode(Panda.SAFETY_ALLOUTPUT)
- receiver.set_safety_mode(Panda.SAFETY_ALLOUTPUT)
+ sender.set_heartbeat_disabled()
# Start transmisson
threading.Thread(target=flood_tx, args=(sender,)).start()
@@ -33,7 +42,7 @@ if __name__ == "__main__":
rx: List[Any] = []
old_len = 0
start_time = time.time()
- while time.time() - start_time < 2 or len(rx) > old_len:
+ while time.time() - start_time < 3 or len(rx) > old_len:
old_len = len(rx)
rx.extend(receiver.can_recv())
print(f"Received {len(rx)} messages")
