FilesExpand file tree

 develop

/

serial_com.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

executable file

·

81 lines (62 loc) · 2.24 KB

 develop

/

serial_com.py

Copy path

Top

File metadata and controls

• Code
• Blame

executable file

·

81 lines (62 loc) · 2.24 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

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

#!/usr/bin/env python

"""

This example sends every second a messages over the serial interface and also

receives incoming messages.

python3 -m examples.serial_com

Expects two serial ports (/dev/ttyS10 and /dev/ttyS11) connected to each other:

Linux:

To connect two ports use socat.

sudo apt-get install socat

sudo socat PTY,link=/dev/ttyS10 PTY,link=/dev/ttyS11

Windows:

This example was not tested on Windows. To create and connect virtual

ports on Windows, the following software can be used:

com0com: http://com0com.sourceforge.net/

"""

import time

import threading

import can

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(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)

if rx_msg is not None:

print("rx: {}".format(rx_msg))

print("Stopped receiving messages")

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()