From fab8d0bb4a7704d771bbac79e1cf94c6469f8888 Mon Sep 17 00:00:00 2001
From: Radek <radek@comma.ai>
Date: Wed, 2 Oct 2024 20:53:32 +0100
Subject: [PATCH] CAN print improvement (#2052)

* Jungle spam can.

* Print all busses by default.  Prettier print.

* can_health printer

* colors on can error
---
 board/jungle/scripts/can_health.py  | 20 ++++++++++++++++++--
 board/jungle/scripts/can_printer.py | 17 +++++++++++------
 board/jungle/scripts/spam_can.py    | 20 ++++++++++++++++++++
 tests/can_health.py                 | 29 +++++++++++++++++++++++++++++
 tests/can_printer.py                | 20 ++++++++++----------
 5 files changed, 88 insertions(+), 18 deletions(-)
 create mode 100755 board/jungle/scripts/spam_can.py
 create mode 100755 tests/can_health.py

diff --git a/board/jungle/scripts/can_health.py b/board/jungle/scripts/can_health.py
index ff068b5baa..47efbb2e67 100755
--- a/board/jungle/scripts/can_health.py
+++ b/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)
diff --git a/board/jungle/scripts/can_printer.py b/board/jungle/scripts/can_printer.py
index f620d6f122..05cc802aa6 100755
--- a/board/jungle/scripts/can_printer.py
+++ b/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()
 
diff --git a/board/jungle/scripts/spam_can.py b/board/jungle/scripts/spam_can.py
new file mode 100755
index 0000000000..c3d798b473
--- /dev/null
+++ b/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)
diff --git a/tests/can_health.py b/tests/can_health.py
new file mode 100755
index 0000000000..26de9cdab1
--- /dev/null
+++ b/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)
diff --git a/tests/can_printer.py b/tests/can_printer.py
index 70a18a61bd..791f8e75dc 100755
--- a/tests/can_printer.py
+++ b/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()