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
|
#!/usr/bin/env python3
"""
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(f"rx: {rx_msg}")
print("Stopped receiving messages")
def main():
"""Controls the sender and receiver."""
with can.Bus(interface="serial", channel="/dev/ttyS10") as server:
with can.Bus(interface="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()
|
