use Notifier context manager in examples

Author: zariiii9003

examples/asyncio_demo.py

@@ -5,10 +5,12 @@
 """
 
 import asyncio
-from typing import List
+from typing import TYPE_CHECKING, List
 
 import can
-from can.notifier import MessageRecipient
+
+if TYPE_CHECKING:
+    from can.notifier import MessageRecipient
 
 
 def print_message(msg: can.Message) -> None:
@@ -31,26 +33,22 @@ async def main() -> None:
             logger,  # Regular Listener object
         ]
         # Create Notifier with an explicit loop to use for scheduling of callbacks
-        loop = asyncio.get_running_loop()
-        notifier = can.Notifier(bus, listeners, loop=loop)
-        # Start sending first message
-        bus.send(can.Message(arbitration_id=0))
-
-        print("Bouncing 10 messages...")
-        for _ in range(10):
-            # Wait for next message from AsyncBufferedReader
-            msg = await reader.get_message()
-            # Delay response
-            await asyncio.sleep(0.5)
-            msg.arbitration_id += 1
-            bus.send(msg)
-
-        # Wait for last message to arrive
-        await reader.get_message()
-        print("Done!")
-
-        # Clean-up
-        notifier.stop()
+        with can.Notifier(bus, listeners, loop=asyncio.get_running_loop()):
+            # Start sending first message
+            bus.send(can.Message(arbitration_id=0))
+
+            print("Bouncing 10 messages...")
+            for _ in range(10):
+                # Wait for next message from AsyncBufferedReader
+                msg = await reader.get_message()
+                # Delay response
+                await asyncio.sleep(0.5)
+                msg.arbitration_id += 1
+                bus.send(msg)
+
+            # Wait for last message to arrive
+            await reader.get_message()
+            print("Done!")
 
 
 if __name__ == "__main__":

examples/cyclic_checksum.py

@@ -59,6 +59,5 @@ def compute_xbr_checksum(message: can.Message, counter: int) -> int:
 
 if __name__ == "__main__":
     with can.Bus(channel=0, interface="virtual", receive_own_messages=True) as _bus:
-        notifier = can.Notifier(bus=_bus, listeners=[print])
-        cyclic_checksum_send(_bus)
-        notifier.stop()
+        with can.Notifier(bus=_bus, listeners=[print]):
+            cyclic_checksum_send(_bus)

examples/print_notifier.py

@@ -8,14 +8,13 @@
 def main():
     with can.Bus(interface="virtual", receive_own_messages=True) as bus:
         print_listener = can.Printer()
-        notifier = 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)
-        notifier.stop()
+        with can.Notifier(bus, listeners=[print_listener]):
+            # using Notifier as a context manager automatically calls `Notifier.stop()`
+            # at the end of the `with` block
+            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__":

examples/send_multiple.py

@@ -4,8 +4,8 @@
 This demo creates multiple processes of producers to spam a socketcan bus.
 """
 
-from time import sleep
 from concurrent.futures import ProcessPoolExecutor
+from time import sleep
 
 import can
 

examples/serial_com.py

@@ -18,8 +18,8 @@
         com0com: http://com0com.sourceforge.net/
 """
 
-import time
 import threading
+import time
 
 import can
 

examples/vcan_filtered.py

@@ -18,14 +18,11 @@ def main():
         # print all incoming messages, which 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)
-        notifier.stop()
+        with 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__":

pyproject.toml

@@ -167,6 +167,7 @@ ignore = [
 ]
 "can/logger.py" = ["T20"]  # flake8-print
 "can/player.py" = ["T20"]  # flake8-print
+"examples/*" = ["T20"]  # flake8-print
 
 [tool.ruff.lint.isort]
 known-first-party = ["can"]