-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathTestSerial2.py
More file actions
156 lines (127 loc) · 4.69 KB
/
Copy pathTestSerial2.py
File metadata and controls
156 lines (127 loc) · 4.69 KB
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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
import serial
# Module sys has to be imported:
import sys
import pyudev
import time, os
from serial import Serial
from serial.serialutil import SerialException
import serial.tools.list_ports
mySerial = serial.Serial()
def _send_command():
global mySerial
try:
if mySerial.inWaiting() != 0:
raise Exception(
'Input buffer must be empty before sending command. Currently %s bytes in the input buffer.' % self._serial.inWaiting())
cmd = bytes([0xFF, 0x02, 0x02, 0xF0, 0x01, 0x4D, 0x6D, 0x0A, 0x03])
mySerial.write(cmd)
print("command sent")
except (SerialException, OSError) as msg:
print('Could not send command: %s' % msg)
return mySerial.in_waiting
def _connect_reader(port):
global mySerial
print("connect")
try:
mySerial = Serial(port, baudrate=38400, timeout=5)
except (SerialException, OSError):
print("Could not open port '%s'" % port)
# flush possibly available input
try:
mySerial.flushInput()
except (SerialException, OSError) as msg:
# This happens if the serial port is not ready for
# whatever reason (eg. there is no real device behind this device node).
print("Could not flush port '%s'" % msg)
try:
# try at 38400 baud, extended protocol
_send_command()
except:
try:
print("exc send_command")
mySerial.baudrate = 4800
except (SerialException, OSError) as msg:
raise Exception('Could not set port speed to 4800: %s' % msg)
try:
_send_command()
except:
print('This module only works with BSM7/8 stations: %s')
time.sleep(0.5)
print("waiting in")
while mySerial.in_waiting != 0:
readBytes = mySerial.read()
print(readBytes)
while True:
mylist = serial.tools.list_ports.grep('10c4:800a|0525:a4aa')
for a in mylist:
print(a.device)
time.sleep(5)
continue
serialPort = None
# https://github.com/dhylands/usb-ser-mon/blob/master/find_port.py
uDevContext = pyudev.Context()
for device in uDevContext.list_devices(subsystem='tty'):
if 'ID_VENDOR_ID' in device:
if device['ID_VENDOR_ID'].lower() == '10c4' and \
device['ID_MODEL_ID'].lower() == '800a':
serialPort = device.device_node
elif device['ID_VENDOR_ID'].lower() == '0525' and \
device['ID_MODEL_ID'].lower() == 'a4aa':
serialPort = device.device_node
if serialPort is not None:
_connect_reader(serialPort)
time.sleep(10)
continue
mySerial.baudrate = 38400
mySerial.port = serialPort
#mySerial.writeTimeout = 1
#mySerial.port = '/dev/ttyUSB3'
#mySerial.port = '/dev/ttyGS0'
#mySerial.timeout = 5
if mySerial.is_open:
print("Serial is open")
else:
print("Serial is not open")
if not mySerial.is_open:
print("Try to open")
mySerial.open()
print("After try to open")
if mySerial.is_open:
print("Serial is now open")
else:
print("Serial is still not open")
print("Check if any bytes waiting")
noOfBytes = mySerial.in_waiting
print("No of bytes: " + str(noOfBytes))
print("ReceiveSIAdapter::Init() serial port open")
msdMode = bytes([0xFF, 0x02, 0x02, 0xF0, 0x01, 0x4D, 0x6D, 0x0A, 0x03])
a = mySerial.write(msdMode)
#mySerial.flushOutput()
#mySerial.flushInput()
time.sleep(1)
print("written: " + str(a))
print("Check if any bytes out waiting")
#noOfBytes = mySerial.out_waiting
print("No of bytes out: " + str(noOfBytes))
expectedLength = 3
response = bytearray()
startFound = False
while mySerial.in_waiting > 0:
bytesRead = mySerial.read(1)
print("bytesRead: " + str(bytesRead))
if bytesRead[0] == 0x02:
startFound = True
if startFound:
response.append(bytesRead[0])
if len(response) == 3:
expectedLength = response[2] + 6
if len(response) == expectedLength:
break
if len(response) < expectedLength and mySerial.in_waiting == 0:
print("ReceiveSIAdapter::Init() sleep and wait for more bytes")
time.sleep(0.05)
if mySerial.in_waiting == 0:
break
print("ReceiveSIAdapter::Init() received: " + str(response))
time.sleep(10)
#mySerial.close()