FilesExpand file tree

 develop

/

asyncio_demo.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

executable file

·

61 lines (48 loc) · 1.53 KB

 develop

/

asyncio_demo.py

Copy path

Top

File metadata and controls

• Code
• Blame

executable file

·

61 lines (48 loc) · 1.53 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

#!/usr/bin/env python

"""

This example demonstrates how to use async IO with python-can.

"""

import asyncio

import can

def print_message(msg):

"""Regular callback function. Can also be a coroutine."""

print(msg)

async def main():

"""The main function that runs in the loop."""

bus = can.Bus("vcan0", bustype="virtual", receive_own_messages=True)

reader = can.AsyncBufferedReader()

logger = can.Logger("logfile.asc")

listeners = [

print_message, # Callback function

reader, # AsyncBufferedReader() listener

logger, # Regular Listener object

]

# Create Notifier with an explicit loop to use for scheduling of callbacks

loop = asyncio.get_event_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()

bus.shutdown()

if __name__ == "__main__":

try:

# Get the default event loop

LOOP = asyncio.get_event_loop()

# Run until main coroutine finishes

LOOP.run_until_complete(main())

finally:

LOOP.close()

# or on Python 3.7+ simply

# asyncio.run(main())