forked from microsoft/devicescript
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathjdusb.py
More file actions
executable file
·115 lines (99 loc) · 3.57 KB
/
Copy pathjdusb.py
File metadata and controls
executable file
·115 lines (99 loc) · 3.57 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
#!/usr/bin/env python
#
# Command line driver based on:
# (C) 2002-2020 Chris Liechti <cliechti@gmx.net>
# SPDX-License-Identifier: BSD-3-Clause
import sys
import socket
import serial
import serial.threaded
import time
import jdusbproto
if __name__ == '__main__': # noqa
import argparse
parser = argparse.ArgumentParser(
description='JDUSB/serial to tcp:// connector')
parser.add_argument(
'SERIALPORT',
help="serial port name")
parser.add_argument(
'BAUDRATE',
type=int,
nargs='?',
help='set baud rate, default: %(default)s',
default=1500000)
parser.add_argument(
'--develop',
action='store_true',
help='Development mode, prints Python internals on errors',
default=False)
parser.add_argument(
'-c', '--client',
metavar='HOST:PORT',
help='connect to devtools, default: %(default)s',
default='localhost:8082')
args = parser.parse_args()
# connect to serial port
ser = serial.serial_for_url(args.SERIALPORT, do_not_open=True)
ser.baudrate = args.BAUDRATE
sys.stderr.write(
'--- TCP/IP to Serial redirect on {p.name} {p.baudrate},{p.bytesize},{p.parity},{p.stopbits} ---\n'
'--- type Ctrl-C / BREAK to quit\n'.format(p=ser))
try:
ser.open()
except serial.SerialException as e:
sys.stderr.write('Could not open serial port {}: {}\n'.format(ser.name, e))
sys.exit(1)
ser_to_net = jdusbproto.JdUsbProto(ser)
serial_worker = serial.threaded.ReaderThread(ser, ser_to_net)
serial_worker.start()
sleep_time = 0
try:
while True:
time.sleep(sleep_time)
host, port = args.client.split(':')
sys.stderr.write("connecting to {}:{}...\n".format(host, port))
sleep_time = 5
client_socket = socket.socket()
try:
client_socket.connect((host, int(port)))
except socket.error as msg:
sys.stderr.write('WARNING: {}\n'.format(msg))
continue
sys.stderr.write('Connected\n')
client_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
#~ client_socket.settimeout(5)
try:
ser_to_net.socket = client_socket
# enter network <-> serial loop
prev_data = b''
while True:
try:
data = client_socket.recv(1024)
if not data:
break
prev_data += data
while len(prev_data) > 0:
ln = prev_data[0]
if len(prev_data) <= ln:
break
ser_to_net.send_frame(prev_data[1:ln+1])
prev_data = prev_data[ln+1:]
except socket.error as msg:
if args.develop:
raise
sys.stderr.write('ERROR: {}\n'.format(msg))
# probably got disconnected
break
except socket.error as msg:
if args.develop:
raise
sys.stderr.write('ERROR: {}\n'.format(msg))
finally:
ser_to_net.socket = None
sys.stderr.write('Disconnected\n')
client_socket.close()
except KeyboardInterrupt:
pass
sys.stderr.write('\n--- exit ---\n')
serial_worker.stop()