cleanup the examples; they now pass the linter and formatter and always close the bus by using a context manager

Author: felixdivo

examples/asyncio_demo.py

@@ -58,4 +58,4 @@ async def main():
         LOOP.close()
 
     # or on Python 3.7+ simply
-    #asyncio.run(main())
+    # asyncio.run(main())

examples/cyclic.py

@@ -34,13 +34,14 @@ def simple_periodic_send(bus):
 
 
 def limited_periodic_send(bus):
+    """Send using LimitedDurationCyclicSendTaskABC."""
     print("Starting to send a message every 200ms for 1s")
     msg = can.Message(
         arbitration_id=0x12345678, data=[0, 0, 0, 0, 0, 0], is_extended_id=True
     )
     task = bus.send_periodic(msg, 0.20, 1, store_task=False)
     if not isinstance(task, can.LimitedDurationCyclicSendTaskABC):
-        print("This interface doesn't seem to support a ")
+        print("This interface doesn't seem to support LimitedDurationCyclicSendTaskABC")
         task.stop()
         return
 
@@ -53,7 +54,8 @@ def limited_periodic_send(bus):
 
 
 def test_periodic_send_with_modifying_data(bus):
-    print("Starting to send a message every 200ms. Initial data is ones")
+    """Send using ModifiableCyclicTaskABC."""
+    print("Starting to send a message every 200ms. Initial data is four consecutive 1s")
     msg = can.Message(arbitration_id=0x0CF02200, data=[1, 1, 1, 1])
     task = bus.send_periodic(msg, 0.20)
     if not isinstance(task, can.ModifiableCyclicTaskABC):
@@ -106,19 +108,13 @@ def test_periodic_send_with_modifying_data(bus):
 #     print("done")
 
 
-if __name__ == "__main__":
-
+def main():
+    """Test different cyclic sending tasks."""
     reset_msg = can.Message(
         arbitration_id=0x00, data=[0, 0, 0, 0, 0, 0], is_extended_id=False
     )
 
-    for interface, channel in [
-        ("socketcan", "vcan0"),
-        # ('ixxat', 0)
-    ]:
-        print("Carrying out cyclic tests with {} interface".format(interface))
-
-        bus = can.Bus(interface=interface, channel=channel, bitrate=500000)
+    with can.Bus(interface="virtual") as bus:
         bus.send(reset_msg)
 
         simple_periodic_send(bus)
@@ -133,6 +129,8 @@ def test_periodic_send_with_modifying_data(bus):
         # can.rc['interface'] = interface
         # test_dual_rate_periodic_send()
 
-        bus.shutdown()
-
     time.sleep(2)
+
+
+if __name__ == "__main__":
+    main()

examples/cyclic_multiple.py

@@ -131,14 +131,10 @@ def cyclic_multiple_send_modify(bus):
 
 if __name__ == "__main__":
     for interface, channel in [("socketcan", "vcan0")]:
-        print("Carrying out cyclic multiple tests with {} interface".format(interface))
+        print(f"Carrying out cyclic multiple tests with {interface} interface")
 
-        bus = can.Bus(interface=interface, channel=channel, bitrate=500000)
-
-        cyclic_multiple_send(bus)
-
-        cyclic_multiple_send_modify(bus)
-
-        bus.shutdown()
+        with can.Bus(interface=interface, channel=channel, bitrate=500000) as BUS:
+            cyclic_multiple_send(BUS)
+            cyclic_multiple_send_modify(BUS)
 
     time.sleep(2)

examples/receive_all.py

@@ -1,24 +1,33 @@
 #!/usr/bin/env python
 
+"""
+Shows how the receive messages via polling.
+"""
+
 import can
 from can.bus import BusState
 
 
 def receive_all():
+    """Receives all messages and prints them to the console until Ctrl+C is pressed."""
+
+    with can.interface.Bus(
+        bustype="pcan", channel="PCAN_USBBUS1", bitrate=250000
+    ) as bus:
+        # bus = can.interface.Bus(bustype='ixxat', channel=0, bitrate=250000)
+        # bus = can.interface.Bus(bustype='vector', app_name='CANalyzer', channel=0, bitrate=250000)
 
-    bus = can.interface.Bus(bustype="pcan", channel="PCAN_USBBUS1", bitrate=250000)
-    # bus = can.interface.Bus(bustype='ixxat', channel=0, bitrate=250000)
-    # bus = can.interface.Bus(bustype='vector', app_name='CANalyzer', channel=0, bitrate=250000)
+        # set to read-only, only supported on some interfaces
+        bus.state = BusState.PASSIVE
 
-    bus.state = BusState.ACTIVE  # or BusState.PASSIVE
+        try:
+            while True:
+                msg = bus.recv(1)
+                if msg is not None:
+                    print(msg)
 
-    try:
-        while True:
-            msg = bus.recv(1)
-            if msg is not None:
-                print(msg)
-    except KeyboardInterrupt:
-        pass
+        except KeyboardInterrupt:
+            pass  # exit normally
 
 
 if __name__ == "__main__":

examples/send_one.py

@@ -8,27 +8,28 @@
 
 
 def send_one():
+    """Sends a single message."""
 
     # this uses the default configuration (for example from the config file)
     # see https://python-can.readthedocs.io/en/stable/configuration.html
-    bus = can.interface.Bus()
-
-    # Using specific buses works similar:
-    # bus = can.interface.Bus(bustype='socketcan', channel='vcan0', bitrate=250000)
-    # bus = can.interface.Bus(bustype='pcan', channel='PCAN_USBBUS1', bitrate=250000)
-    # bus = can.interface.Bus(bustype='ixxat', channel=0, bitrate=250000)
-    # bus = can.interface.Bus(bustype='vector', app_name='CANalyzer', channel=0, bitrate=250000)
-    # ...
-
-    msg = can.Message(
-        arbitration_id=0xC0FFEE, data=[0, 25, 0, 1, 3, 1, 4, 1], is_extended_id=True
-    )
-
-    try:
-        bus.send(msg)
-        print("Message sent on {}".format(bus.channel_info))
-    except can.CanError:
-        print("Message NOT sent")
+    with can.interface.Bus() as bus:
+
+        # Using specific buses works similar:
+        # bus = can.interface.Bus(bustype='socketcan', channel='vcan0', bitrate=250000)
+        # bus = can.interface.Bus(bustype='pcan', channel='PCAN_USBBUS1', bitrate=250000)
+        # bus = can.interface.Bus(bustype='ixxat', channel=0, bitrate=250000)
+        # bus = can.interface.Bus(bustype='vector', app_name='CANalyzer', channel=0, bitrate=250000)
+        # ...
+
+        msg = can.Message(
+            arbitration_id=0xC0FFEE, data=[0, 25, 0, 1, 3, 1, 4, 1], is_extended_id=True
+        )
+
+        try:
+            bus.send(msg)
+            print(f"Message sent on {bus.channel_info}")
+        except can.CanError:
+            print("Message NOT sent")
 
 
 if __name__ == "__main__":

examples/serial_com.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 
 """
-This example sends every second a messages over the serial interface and also 
+This example sends every second a messages over the serial interface and also
 receives incoming messages.
 
 python3 -m examples.serial_com
@@ -25,17 +25,19 @@
 
 
 def send_cyclic(bus, msg, stop_event):
+    """The loop for sending."""
     print("Start to send a message every 1s")
     start_time = time.time()
     while not stop_event.is_set():
         msg.timestamp = time.time() - start_time
         bus.send(msg)
-        print("tx: {}".format(tx_msg))
+        print(f"tx: {msg}")
         time.sleep(1)
     print("Stopped sending messages")
 
 
 def receive(bus, stop_event):
+    """The loop for receiving."""
     print("Start receiving messages")
     while not stop_event.is_set():
         rx_msg = bus.recv(1)
@@ -44,30 +46,36 @@ def receive(bus, stop_event):
     print("Stopped receiving messages")
 
 
-if __name__ == "__main__":
-    server = can.interface.Bus(bustype="serial", channel="/dev/ttyS10")
-    client = can.interface.Bus(bustype="serial", channel="/dev/ttyS11")
-
-    tx_msg = can.Message(
-        arbitration_id=0x01, data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88]
-    )
-
-    # Thread for sending and receiving messages
-    stop_event = threading.Event()
-    t_send_cyclic = threading.Thread(
-        target=send_cyclic, args=(server, tx_msg, stop_event)
-    )
-    t_receive = threading.Thread(target=receive, args=(client, stop_event))
-    t_receive.start()
-    t_send_cyclic.start()
-
-    try:
-        while True:
-            pass
-    except KeyboardInterrupt:
-        pass
-
-    stop_event.set()
-    server.shutdown()
-    client.shutdown()
+def main():
+    """Controles the sender and receiver."""
+    with can.interface.Bus(bustype="serial", channel="/dev/ttyS10") as server:
+        with can.interface.Bus(bustype="serial", channel="/dev/ttyS11") as client:
+
+            tx_msg = can.Message(
+                arbitration_id=0x01,
+                data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88],
+            )
+
+            # Thread for sending and receiving messages
+            stop_event = threading.Event()
+            t_send_cyclic = threading.Thread(
+                target=send_cyclic, args=(server, tx_msg, stop_event)
+            )
+            t_receive = threading.Thread(target=receive, args=(client, stop_event))
+            t_receive.start()
+            t_send_cyclic.start()
+
+            try:
+                while True:
+                    time.sleep(0)  # yield
+            except KeyboardInterrupt:
+                pass  # exit normally
+
+            stop_event.set()
+            time.sleep(0.5)
+
     print("Stopped script")
+
+
+if __name__ == "__main__":
+    main()

examples/simple_log_converter.py

@@ -2,17 +2,25 @@
 
 """
 Use this to convert .can/.asc files to .log files.
+Can be easily adapted for all sorts of files.
 
-Usage: simpleLogConvert.py sourceLog.asc targetLog.log
+Usage: python3 simple_log_convert.py sourceLog.asc targetLog.log
 """
 
 import sys
 
-import can.io.logger
-import can.io.player
+import can
 
-reader = can.io.player.LogReader(sys.argv[1])
-writer = can.io.logger.Logger(sys.argv[2])
 
-for msg in reader:
-    writer.on_message_received(msg)
+def main():
+    """The transcoder"""
+
+    with can.LogReader(sys.argv[1]) as reader:
+        with can.Logger(sys.argv[2]) as writer:
+
+            for msg in reader:
+                writer.on_message_received(msg)
+
+
+if __name__ == "__main__":
+    main()

examples/vcan_filtered.py

@@ -8,16 +8,25 @@
 
 import can
 
+
+def main():
+    """Send some messages to itself and apply filtering."""
+    with can.Bus(bustype="virtual", receive_own_messages=True) as bus:
+
+        can_filters = [{"can_id": 1, "can_mask": 0xF, "extended": True}]
+        bus.set_filters(can_filters)
+
+        # print all incoming messages, wich includes the ones sent,
+        # since we set receive_own_messages to True
+        # assign to some variable so it does not garbage collected
+        notifier = can.Notifier(bus, [can.Printer()])  # pylint: disable=unused-variable
+
+        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__":
-    bus = can.interface.Bus(
-        bustype="socketcan", channel="vcan0", receive_own_messages=True
-    )
-
-    can_filters = [{"can_id": 1, "can_mask": 0xF, "extended": True}]
-    bus.set_filters(can_filters)
-    notifier = can.Notifier(bus, [can.Printer()])
-    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(10)
+    main()

examples/virtual_can_demo.py

@@ -10,23 +10,22 @@
 import can
 
 
-def producer(id, message_count=16):
+def producer(thread_id, message_count=16):
     """Spam the bus with messages including the data id.
 
     :param int id: the id of the thread/process
     """
-
     with can.Bus(bustype="socketcan", channel="vcan0") as bus:
         for i in range(message_count):
             msg = can.Message(
-                arbitration_id=0x0CF02200 + id, data=[id, i, 0, 1, 3, 1, 4, 1]
+                arbitration_id=0x0CF02200 + id, data=[thread_id, i, 0, 1, 3, 1, 4, 1]
             )
             bus.send(msg)
         sleep(1.0)
 
-    print("Producer #{} finished sending {} messages".format(id, message_count))
+    print(f"Producer #{thread_id} finished sending {message_count} messages")
 
 
 if __name__ == "__main__":
-    with ProcessPoolExecutor(max_workers=4) as executor:
+    with ProcessPoolExecutor() as executor:
         executor.map(producer, range(5))