CAN print improvement

board/jungle/scripts/can_health.py

@@ -1,13 +1,29 @@
 #!/usr/bin/env python3
-
 import time
+import re
 from panda import PandaJungle
 
+RED = '\033[91m'
+GREEN = '\033[92m'
+
+def colorize_errors(value):
+    if isinstance(value, str):
+        if re.search(r'(?i)No error', value):
+            return f'{GREEN}{value}\033[0m'
+        elif re.search(r'(?i)(?<!No error\s)(err|error)', value):
+            return f'{RED}{value}\033[0m'
+    return str(value)
+
 if __name__ == "__main__":
   jungle = PandaJungle()
 
   while True:
+    print(chr(27) + "[2J") # clear screen
+    print("Connected to " + ("internal panda" if jungle.is_internal() else "External panda") + f" id: {jungle.get_serial()[0]}: {jungle.get_version()}")
     for bus in range(3):
-      print(bus, jungle.can_health(bus))
+      print(f"\nBus {bus}:")
+      health = jungle.can_health(bus)
+      for key, value in health.items():
+        print(f"{key}: {colorize_errors(value)}  ", end=" ")
     print()
     time.sleep(1)

board/jungle/scripts/can_printer.py

@@ -12,22 +12,27 @@ def sec_since_boot():
 
 def can_printer():
   p = PandaJungle()
+  print(f"Connected to: {p._serial}: {p.get_version()}")
+  time.sleep(1)
+
+  p.can_clear(0xFFFF)
 
   start = sec_since_boot()
   lp = sec_since_boot()
   msgs = defaultdict(list)
-  canbus = int(os.getenv("CAN", "0"))
+  canbus = os.getenv("CAN")
   while True:
     can_recv = p.can_recv()
-    for address, dat, src  in can_recv:
-      if src == canbus:
-        msgs[address].append(dat)
+    for address,dat, src in can_recv:
+      if canbus is None or str(src) == canbus:
+        msgs[address].append((dat, src))
 
     if sec_since_boot() - lp > 0.1:
       dd = chr(27) + "[2J"
       dd += "%5.2f\n" % (sec_since_boot() - start)
-      for k,v in sorted(zip(list(msgs.keys()), [binascii.hexlify(x[-1]) for x in list(msgs.values())], strict=True)):
-        dd += "%s(%6d) %s\n" % ("%04X(%4d)" % (k,k),len(msgs[k]), v)
+      for k, v in sorted(msgs.items()):
+        last_msg, last_src = v[-1]
+        dd += "%d: %s(%6d): %s\n" % (last_src, "%04X(%4d)" % (k, k), len(v), binascii.hexlify(last_msg).decode())
       print(dd)
       lp = sec_since_boot()
 

board/jungle/scripts/spam_can.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+import os
+import random
+from panda import PandaJungle
+
+def get_test_string():
+  return b"test" + os.urandom(10)
+
+if __name__ == "__main__":
+  p = PandaJungle()
+
+  p.set_safety_mode(PandaJungle.SAFETY_ALLOUTPUT)
+
+  print("Spamming all buses...")
+  while True:
+    at = random.randint(1, 2000)
+    st = get_test_string()[0:8]
+    bus = random.randint(0, 2)
+    p.can_send(at, st, bus)
+    # print("Sent message on bus: ", bus)

tests/can_health.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+import time
+import re
+from panda import Panda
+
+RED = '\033[91m'
+GREEN = '\033[92m'
+
+def colorize_errors(value):
+    if isinstance(value, str):
+        if re.search(r'(?i)No error', value):
+            return f'{GREEN}{value}\033[0m'
+        elif re.search(r'(?i)(?<!No error\s)(err|error)', value):
+            return f'{RED}{value}\033[0m'
+    return str(value)
+
+if __name__ == "__main__":
+
+  panda = Panda()
+  while True:
+    print(chr(27) + "[2J") # clear screen
+    print("Connected to " + ("internal panda" if panda.is_internal() else "External panda") + f" id: {panda.get_serial()[0]}: {panda.get_version()}")
+    for bus in range(3):
+      print(f"\nBus {bus}:")
+      health = panda.can_health(bus)
+      for key, value in health.items():
+        print(f"{key}: {colorize_errors(value)}  ", end=" ")
+    print()
+    time.sleep(1)

tests/can_printer.py

@@ -12,28 +12,28 @@ def sec_since_boot():
 
 def can_printer():
   p = Panda()
+  print(f"Connected to id: {p.get_serial()[0]}: {p.get_version()}")
+  time.sleep(1)
+
+  p.can_clear(0xFFFF)
   p.set_safety_mode(Panda.SAFETY_ALLOUTPUT)
 
   start = sec_since_boot()
   lp = sec_since_boot()
   msgs = defaultdict(list)
-
-  canbus = int(os.getenv("CAN", "0"))
-  if canbus == 3:
-    canbus = 1
-    p.set_obd(True)
-
+  canbus = os.getenv("CAN")
   while True:
     can_recv = p.can_recv()
     for address, dat, src in can_recv:
-      if src == canbus:
-        msgs[address].append(dat)
+      if canbus is None or str(src) == canbus:
+        msgs[address].append((dat, src))
 
     if sec_since_boot() - lp > 0.1:
       dd = chr(27) + "[2J"
       dd += "%5.2f\n" % (sec_since_boot() - start)
-      for k, v in sorted(zip(list(msgs.keys()), [binascii.hexlify(x[-1]) for x in list(msgs.values())], strict=True)):
-        dd += "%s(%6d) %s\n" % ("%04X(%4d)" % (k, k), len(msgs[k]), v)
+      for k, v in sorted(msgs.items()):
+        last_msg, last_src = v[-1]
+        dd += "%d: %s(%6d): %s\n" % (last_src, "%04X(%4d)" % (k, k), len(v), binascii.hexlify(last_msg).decode())
       print(dd)
       lp = sec_since_boot()