Canbus poc

core.py

@@ -0,0 +1,19 @@
+import can
+import time
+
+def main():
+    time.sleep(5)
+    bus = can.interface.Bus(channel="can0", bustype="socketcan", loopback=True)
+    notifier = can.Notifier(bus, [can.Printer()])
+
+    try:
+        # Simulate doing some work
+        time.sleep(10)
+    finally:
+        # Ensure the notifier is stopped first
+        notifier.stop()
+        # Then shut down the bus
+        bus.shutdown()
+
+if __name__ == "__main__":
+    main()

custom_listener.py

@@ -0,0 +1,14 @@
+import can
+import time
+bus = can.interface.Bus(channel="can0", bustype="socketcan", receive_own_messages=True)
+msg = can.Message(arbitration_id=0x7de, data=[0, 25, 0, 1, 3, 1, 4, 1], is_extended_id=False)
+
+try:
+    while True:
+        time.sleep(1)
+        bus.send(msg)
+        print("Message sent")
+except can.CanError as e:
+    print("Message failed to send", e)
+
+bus.shutdown()

mthread_listener.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python
+
+import time
+import can
+
+
+def main():
+    with can.Bus(receive_own_messages=True) as bus:
+        print_listener = can.Printer()
+        can.Notifier(bus, [print_listener])
+
+        bus.send(can.Message(arbitration_id=1, is_extended_id=True))
+        bus.send(can.Message(arbitration_id=2, is_extended_id=True))
+        bus.send(can.Message(arbitration_id=1, is_extended_id=False))
+
+        time.sleep(1.0)
+
+
+if __name__ == "__main__":
+    main()