1
#!/usr/bin/env python3
2
#
3
# This file is part of the MicroPython project, http://micropython.org/
4
#
5
# The MIT License (MIT)
6
#
7
# Copyright (c) 2014-2021 Damien P. George
8
# Copyright (c) 2017 Paul Sokolovsky
9
#
10
# Permission is hereby granted, free of charge, to any person obtaining a copy
11
# of this software and associated documentation files (the "Software"), to deal
12
# in the Software without restriction, including without limitation the rights
13
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
14
# copies of the Software, and to permit persons to whom the Software is
15
# furnished to do so, subject to the following conditions:
16
#
17
# The above copyright notice and this permission notice shall be included in
18
# all copies or substantial portions of the Software.
19
#
20
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
21
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
22
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
23
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
24
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
25
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
26
# THE SOFTWARE.
27

28
"""
29
pyboard interface
30

31
This module provides the Pyboard class, used to communicate with and
32
control a MicroPython device over a communication channel. Both real
33
boards and emulated devices (e.g. running in QEMU) are supported.
34
Various communication channels are supported, including a serial
35
connection, telnet-style network connection, external process
36
connection.
37

38
Example usage:
39

40
    import pyboard
41
    pyb = pyboard.Pyboard('/dev/ttyACM0')
42

43
Or:
44

45
    pyb = pyboard.Pyboard('192.168.1.1')
46

47
Then:
48

49
    pyb.enter_raw_repl()
50
    pyb.exec('import pyb')
51
    pyb.exec('pyb.LED(1).on()')
52
    pyb.exit_raw_repl()
53

54
Note: if using Python2 then pyb.exec must be written as pyb.exec_.
55
To run a script from the local machine on the board and print out the results:
56

57
    import pyboard
58
    pyboard.execfile('test.py', device='/dev/ttyACM0')
59

60
This script can also be run directly.  To execute a local script, use:
61

62
    ./pyboard.py test.py
63

64
Or:
65

66
    python pyboard.py test.py
67

68
"""
69

70
import ast
71
import errno
72
import os
73
import struct
74
import sys
75
import time
76

77
from collections import namedtuple
78

79
try:
80
    stdout = sys.stdout.buffer
81
except AttributeError:
82
    # Python2 doesn't have buffer attr
83
    stdout = sys.stdout
84

85

86
def stdout_write_bytes(b):
87
    b = b.replace(b"\x04", b"")
88
    stdout.write(b)
89
    stdout.flush()
90

91

92
class PyboardError(Exception):
93
    def convert(self, info):
94
        if len(self.args) >= 3:
95
            if b"OSError" in self.args[2] and b"ENOENT" in self.args[2]:
96
                return OSError(errno.ENOENT, info)
97

98
        return self
99

100

101
listdir_result = namedtuple("dir_result", ["name", "st_mode", "st_ino", "st_size"])
102

103

104
class TelnetToSerial:
105
    def __init__(self, ip, user, password, read_timeout=None):
106
        self.tn = None
107
        import telnetlib
108

109
        self.tn = telnetlib.Telnet(ip, timeout=15)
110
        self.read_timeout = read_timeout
111
        if b"Login as:" in self.tn.read_until(b"Login as:", timeout=read_timeout):
112
            self.tn.write(bytes(user, "ascii") + b"\r\n")
113

114
            if b"Password:" in self.tn.read_until(b"Password:", timeout=read_timeout):
115
                # needed because of internal implementation details of the telnet server
116
                time.sleep(0.2)
117
                self.tn.write(bytes(password, "ascii") + b"\r\n")
118

119
                if b"for more information." in self.tn.read_until(
120
                    b'Type "help()" for more information.', timeout=read_timeout
121
                ):
122
                    # login successful
123
                    from collections import deque
124

125
                    self.fifo = deque()
126
                    return
127

128
        raise PyboardError("Failed to establish a telnet connection with the board")
129

130
    def __del__(self):
131
        self.close()
132

133
    def close(self):
134
        if self.tn:
135
            self.tn.close()
136

137
    def read(self, size=1):
138
        while len(self.fifo) < size:
139
            timeout_count = 0
140
            data = self.tn.read_eager()
141
            if len(data):
142
                self.fifo.extend(data)
143
                timeout_count = 0
144
            else:
145
                time.sleep(0.25)
146
                if self.read_timeout is not None and timeout_count > 4 * self.read_timeout:
147
                    break
148
                timeout_count += 1
149

150
        data = b""
151
        while len(data) < size and len(self.fifo) > 0:
152
            data += bytes([self.fifo.popleft()])
153
        return data
154

155
    def write(self, data):
156
        self.tn.write(data)
157
        return len(data)
158

159
    def inWaiting(self):
160
        n_waiting = len(self.fifo)
161
        if not n_waiting:
162
            data = self.tn.read_eager()
163
            self.fifo.extend(data)
164
            return len(data)
165
        else:
166
            return n_waiting
167

168

169
class ProcessToSerial:
170
    "Execute a process and emulate serial connection using its stdin/stdout."
171

172
    def __init__(self, cmd):
173
        import subprocess
174

175
        self.subp = subprocess.Popen(
176
            cmd,
177
            bufsize=0,
178
            shell=True,
179
            preexec_fn=os.setsid,
180
            stdin=subprocess.PIPE,
181
            stdout=subprocess.PIPE,
182
        )
183

184
        # Initially was implemented with selectors, but that adds Python3
185
        # dependency. However, there can be race conditions communicating
186
        # with a particular child process (like QEMU), and selectors may
187
        # still work better in that case, so left inplace for now.
188
        #
189
        # import selectors
190
        # self.sel = selectors.DefaultSelector()
191
        # self.sel.register(self.subp.stdout, selectors.EVENT_READ)
192

193
        import select
194

195
        self.poll = select.poll()
196
        self.poll.register(self.subp.stdout.fileno())
197

198
    def close(self):
199
        import signal
200

201
        os.killpg(os.getpgid(self.subp.pid), signal.SIGTERM)
202

203
    def read(self, size=1):
204
        data = b""
205
        while len(data) < size:
206
            data += self.subp.stdout.read(size - len(data))
207
        return data
208

209
    def write(self, data):
210
        self.subp.stdin.write(data)
211
        return len(data)
212

213
    def inWaiting(self):
214
        # res = self.sel.select(0)
215
        res = self.poll.poll(0)
216
        if res:
217
            return 1
218
        return 0
219

220

221
class ProcessPtyToTerminal:
222
    """Execute a process which creates a PTY and prints slave PTY as
223
    first line of its output, and emulate serial connection using
224
    this PTY."""
225

226
    def __init__(self, cmd):
227
        import subprocess
228
        import re
229
        import serial
230

231
        self.subp = subprocess.Popen(
232
            cmd.split(),
233
            bufsize=0,
234
            shell=False,
235
            preexec_fn=os.setsid,
236
            stdin=subprocess.PIPE,
237
            stdout=subprocess.PIPE,
238
            stderr=subprocess.STDOUT,
239
        )
240
        pty_line = self.subp.stdout.readline().decode("utf-8")
241
        m = re.search(r"/dev/pts/[0-9]+", pty_line)
242
        if not m:
243
            print("Error: unable to find PTY device in startup line:", pty_line)
244
            self.close()
245
            sys.exit(1)
246
        pty = m.group()
247
        # Compensate for some boards taking a bit longer to start
248
        time.sleep(0.1)
249
        # rtscts, dsrdtr params are to workaround pyserial bug:
250
        # http://stackoverflow.com/questions/34831131/pyserial-does-not-play-well-with-virtual-port
251
        self.serial = serial.Serial(pty, interCharTimeout=1, rtscts=True, dsrdtr=True)
252

253
    def close(self):
254
        import signal
255

256
        os.killpg(os.getpgid(self.subp.pid), signal.SIGTERM)
257

258
    def read(self, size=1):
259
        return self.serial.read(size)
260

261
    def write(self, data):
262
        return self.serial.write(data)
263

264
    def inWaiting(self):
265
        return self.serial.inWaiting()
266

267

268
class Pyboard:
269
    def __init__(
270
        self,
271
        device,
272
        baudrate=115200,
273
        user="micro",
274
        password="python",
275
        wait=0,
276
        exclusive=True,
277
        timeout=None,
278
        write_timeout=5,
279
    ):
280
        self.in_raw_repl = False
281
        self.use_raw_paste = True
282
        if device.startswith("exec:"):
283
            self.serial = ProcessToSerial(device[len("exec:") :])
284
        elif device.startswith("execpty:"):
285
            self.serial = ProcessPtyToTerminal(device[len("qemupty:") :])
286
        elif device and device[0].isdigit() and device[-1].isdigit() and device.count(".") == 3:
287
            # device looks like an IP address
288
            self.serial = TelnetToSerial(device, user, password, read_timeout=10)
289
        else:
290
            import serial
291
            import serial.tools.list_ports
292

293
            # Set options, and exclusive if pyserial supports it
294
            serial_kwargs = {
295
                "baudrate": baudrate,
296
                "timeout": timeout,
297
                "write_timeout": write_timeout,
298
                "interCharTimeout": 1,
299
            }
300
            if serial.__version__ >= "3.3":
301
                serial_kwargs["exclusive"] = exclusive
302

303
            delayed = False
304
            for attempt in range(wait + 1):
305
                try:
306
                    if os.name == "nt":
307
                        self.serial = serial.Serial(**serial_kwargs)
308
                        self.serial.port = device
309
                        portinfo = list(serial.tools.list_ports.grep(device))  # type: ignore
310
                        if portinfo and portinfo[0].manufacturer != "Microsoft":
311
                            # ESP8266/ESP32 boards use RTS/CTS for flashing and boot mode selection.
312
                            # DTR False: to avoid using the reset button will hang the MCU in bootloader mode
313
                            # RTS False: to prevent pulses on rts on serial.close() that would POWERON_RESET an ESPxx
314
                            self.serial.dtr = False  # DTR False = gpio0 High = Normal boot
315
                            self.serial.rts = False  # RTS False = EN High = MCU enabled
316
                        self.serial.open()
317
                    else:
318
                        self.serial = serial.Serial(device, **serial_kwargs)
319
                    break
320
                except (OSError, IOError):  # Py2 and Py3 have different errors
321
                    if wait == 0:
322
                        continue
323
                    if attempt == 0:
324
                        sys.stdout.write("Waiting {} seconds for pyboard ".format(wait))
325
                        delayed = True
326
                time.sleep(1)
327
                sys.stdout.write(".")
328
                sys.stdout.flush()
329
            else:
330
                if delayed:
331
                    print("")
332
                raise PyboardError("failed to access " + device)
333
            if delayed:
334
                print("")
335

336
    def close(self):
337
        self.serial.close()
338

339
    def read_until(
340
        self, min_num_bytes, ending, timeout=10, data_consumer=None, timeout_overall=None
341
    ):
342
        """
343
        min_num_bytes: Obsolete.
344
        ending: Return if 'ending' matches.
345
        timeout [s]: Return if timeout between characters. None: Infinite timeout.
346
        timeout_overall [s]: Return not later than timeout_overall. None: Infinite timeout.
347
        data_consumer: Use callback for incoming characters.
348
            If data_consumer is used then data is not accumulated and the ending must be 1 byte long
349

350
        It is not visible to the caller why the function returned. It could be ending or timeout.
351
        """
352
        assert data_consumer is None or len(ending) == 1
353
        assert isinstance(timeout, (type(None), int, float))
354
        assert isinstance(timeout_overall, (type(None), int, float))
355

356
        data = b""
357
        begin_overall_s = begin_char_s = time.monotonic()
358
        while True:
359
            if data.endswith(ending):
360
                break
361
            elif self.serial.inWaiting() > 0:
362
                new_data = self.serial.read(1)
363
                if data_consumer:
364
                    data_consumer(new_data)
365
                    data = new_data
366
                else:
367
                    data = data + new_data
368
                begin_char_s = time.monotonic()
369
            else:
370
                if timeout is not None and time.monotonic() >= begin_char_s + timeout:
371
                    break
372
                if (
373
                    timeout_overall is not None
374
                    and time.monotonic() >= begin_overall_s + timeout_overall
375
                ):
376
                    break
377
                time.sleep(0.01)
378
        return data
379

380
    def enter_raw_repl(self, soft_reset=True, timeout_overall=10):
381
        try:
382
            self._enter_raw_repl_unprotected(soft_reset, timeout_overall)
383
        except OSError as er:
384
            raise PyboardError("could not enter raw repl: {}".format(er))
385

386
    def _enter_raw_repl_unprotected(self, soft_reset, timeout_overall):
387
        self.serial.write(b"\r\x03")  # ctrl-C: interrupt any running program
388

389
        # flush input (without relying on serial.flushInput())
390
        n = self.serial.inWaiting()
391
        while n > 0:
392
            self.serial.read(n)
393
            n = self.serial.inWaiting()
394

395
        self.serial.write(b"\r\x01")  # ctrl-A: enter raw REPL
396

397
        if soft_reset:
398
            data = self.read_until(
399
                1, b"raw REPL; CTRL-B to exit\r\n>", timeout_overall=timeout_overall
400
            )
401
            if not data.endswith(b"raw REPL; CTRL-B to exit\r\n>"):
402
                print(data)
403
                raise PyboardError("could not enter raw repl")
404

405
            self.serial.write(b"\x04")  # ctrl-D: soft reset
406

407
            # Waiting for "soft reboot" independently to "raw REPL" (done below)
408
            # allows boot.py to print, which will show up after "soft reboot"
409
            # and before "raw REPL".
410
            data = self.read_until(1, b"soft reboot\r\n", timeout_overall=timeout_overall)
411
            if not data.endswith(b"soft reboot\r\n"):
412
                print(data)
413
                raise PyboardError("could not enter raw repl")
414

415
        data = self.read_until(1, b"raw REPL; CTRL-B to exit\r\n", timeout_overall=timeout_overall)
416
        if not data.endswith(b"raw REPL; CTRL-B to exit\r\n"):
417
            print(data)
418
            raise PyboardError("could not enter raw repl")
419

420
        self.in_raw_repl = True
421

422
    def exit_raw_repl(self):
423
        self.serial.write(b"\r\x02")  # ctrl-B: enter friendly REPL
424
        self.in_raw_repl = False
425

426
    def follow(self, timeout, data_consumer=None):
427
        # wait for normal output
428
        data = self.read_until(1, b"\x04", timeout=timeout, data_consumer=data_consumer)
429
        if not data.endswith(b"\x04"):
430
            raise PyboardError("timeout waiting for first EOF reception")
431
        data = data[:-1]
432

433
        # wait for error output
434
        data_err = self.read_until(1, b"\x04", timeout=timeout)
435
        if not data_err.endswith(b"\x04"):
436
            raise PyboardError("timeout waiting for second EOF reception")
437
        data_err = data_err[:-1]
438

439
        # return normal and error output
440
        return data, data_err
441

442
    def raw_paste_write(self, command_bytes):
443
        # Read initial header, with window size.
444
        data = self.serial.read(2)
445
        window_size = struct.unpack("<H", data)[0]
446
        window_remain = window_size
447

448
        # Write out the command_bytes data.
449
        i = 0
450
        while i < len(command_bytes):
451
            while window_remain == 0 or self.serial.inWaiting():
452
                data = self.serial.read(1)
453
                if data == b"\x01":
454
                    # Device indicated that a new window of data can be sent.
455
                    window_remain += window_size
456
                elif data == b"\x04":
457
                    # Device indicated abrupt end.  Acknowledge it and finish.
458
                    self.serial.write(b"\x04")
459
                    return
460
                else:
461
                    # Unexpected data from device.
462
                    raise PyboardError("unexpected read during raw paste: {}".format(data))
463
            # Send out as much data as possible that fits within the allowed window.
464
            b = command_bytes[i : min(i + window_remain, len(command_bytes))]
465
            self.serial.write(b)
466
            window_remain -= len(b)
467
            i += len(b)
468

469
        # Indicate end of data.
470
        self.serial.write(b"\x04")
471

472
        # Wait for device to acknowledge end of data.
473
        data = self.read_until(1, b"\x04")
474
        if not data.endswith(b"\x04"):
475
            raise PyboardError("could not complete raw paste: {}".format(data))
476

477
    def exec_raw_no_follow(self, command):
478
        if isinstance(command, bytes):
479
            command_bytes = command
480
        else:
481
            command_bytes = bytes(command, encoding="utf8")
482

483
        # check we have a prompt
484
        data = self.read_until(1, b">")
485
        if not data.endswith(b">"):
486
            raise PyboardError("could not enter raw repl")
487

488
        if self.use_raw_paste:
489
            # Try to enter raw-paste mode.
490
            self.serial.write(b"\x05A\x01")
491
            data = self.serial.read(2)
492
            if data == b"R\x00":
493
                # Device understood raw-paste command but doesn't support it.
494
                pass
495
            elif data == b"R\x01":
496
                # Device supports raw-paste mode, write out the command using this mode.
497
                return self.raw_paste_write(command_bytes)
498
            else:
499
                # Device doesn't support raw-paste, fall back to normal raw REPL.
500
                data = self.read_until(1, b"w REPL; CTRL-B to exit\r\n>")
501
                if not data.endswith(b"w REPL; CTRL-B to exit\r\n>"):
502
                    print(data)
503
                    raise PyboardError("could not enter raw repl")
504
            # Don't try to use raw-paste mode again for this connection.
505
            self.use_raw_paste = False
506

507
        # Write command using standard raw REPL, 256 bytes every 10ms.
508
        for i in range(0, len(command_bytes), 256):
509
            self.serial.write(command_bytes[i : min(i + 256, len(command_bytes))])
510
            time.sleep(0.01)
511
        self.serial.write(b"\x04")
512

513
        # check if we could exec command
514
        data = self.serial.read(2)
515
        if data != b"OK":
516
            raise PyboardError("could not exec command (response: %r)" % data)
517

518
    def exec_raw(self, command, timeout=10, data_consumer=None):
519
        self.exec_raw_no_follow(command)
520
        return self.follow(timeout, data_consumer)
521

522
    def eval(self, expression, parse=False):
523
        if parse:
524
            ret = self.exec_("print(repr({}))".format(expression))
525
            ret = ret.strip()
526
            return ast.literal_eval(ret.decode())
527
        else:
528
            ret = self.exec_("print({})".format(expression))
529
            ret = ret.strip()
530
            return ret
531

532
    # In Python3, call as pyboard.exec(), see the setattr call below.
533
    def exec_(self, command, data_consumer=None):
534
        ret, ret_err = self.exec_raw(command, data_consumer=data_consumer)
535
        if ret_err:
536
            raise PyboardError("exception", ret, ret_err)
537
        return ret
538

539
    def execfile(self, filename):
540
        with open(filename, "rb") as f:
541
            pyfile = f.read()
542
        return self.exec_(pyfile)
543

544
    def get_time(self):
545
        t = str(self.eval("machine.RTC().datetime()"), encoding="utf8")[1:-1].split(", ")
546
        return int(t[4]) * 3600 + int(t[5]) * 60 + int(t[6])
547

548
    def fs_exists(self, src):
549
        try:
550
            self.exec_("import os\nos.stat(%s)" % (("'%s'" % src) if src else ""))
551
            return True
552
        except PyboardError:
553
            return False
554

555
    def fs_ls(self, src):
556
        cmd = (
557
            "import os\nfor f in os.ilistdir(%s):\n"
558
            " print('{:12} {}{}'.format(f[3]if len(f)>3 else 0,f[0],'/'if f[1]&0x4000 else ''))"
559
            % (("'%s'" % src) if src else "")
560
        )
561
        self.exec_(cmd, data_consumer=stdout_write_bytes)
562

563
    def fs_listdir(self, src=""):
564
        buf = bytearray()
565

566
        def repr_consumer(b):
567
            buf.extend(b.replace(b"\x04", b""))
568

569
        cmd = "import os\nfor f in os.ilistdir(%s):\n print(repr(f), end=',')" % (
570
            ("'%s'" % src) if src else ""
571
        )
572
        try:
573
            buf.extend(b"[")
574
            self.exec_(cmd, data_consumer=repr_consumer)
575
            buf.extend(b"]")
576
        except PyboardError as e:
577
            raise e.convert(src)
578

579
        return [
580
            listdir_result(*f) if len(f) == 4 else listdir_result(*(f + (0,)))
581
            for f in ast.literal_eval(buf.decode())
582
        ]
583

584
    def fs_stat(self, src):
585
        try:
586
            self.exec_("import os")
587
            return os.stat_result(self.eval("os.stat(%s)" % ("'%s'" % src), parse=True))
588
        except PyboardError as e:
589
            raise e.convert(src)
590

591
    def fs_cat(self, src, chunk_size=256):
592
        cmd = (
593
            "with open('%s') as f:\n while 1:\n"
594
            "  b=f.read(%u)\n  if not b:break\n  print(b,end='')" % (src, chunk_size)
595
        )
596
        self.exec_(cmd, data_consumer=stdout_write_bytes)
597

598
    def fs_readfile(self, src, chunk_size=256):
599
        buf = bytearray()
600

601
        def repr_consumer(b):
602
            buf.extend(b.replace(b"\x04", b""))
603

604
        cmd = (
605
            "with open('%s', 'rb') as f:\n while 1:\n"
606
            "  b=f.read(%u)\n  if not b:break\n  print(b,end='')" % (src, chunk_size)
607
        )
608
        try:
609
            self.exec_(cmd, data_consumer=repr_consumer)
610
        except PyboardError as e:
611
            raise e.convert(src)
612
        return ast.literal_eval(buf.decode())
613

614
    def fs_writefile(self, dest, data, chunk_size=256):
615
        self.exec_("f=open('%s','wb')\nw=f.write" % dest)
616
        while data:
617
            chunk = data[:chunk_size]
618
            self.exec_("w(" + repr(chunk) + ")")
619
            data = data[len(chunk) :]
620
        self.exec_("f.close()")
621

622
    def fs_cp(self, src, dest, chunk_size=256, progress_callback=None):
623
        if progress_callback:
624
            src_size = self.fs_stat(src).st_size
625
            written = 0
626
        self.exec_("fr=open('%s','rb')\nr=fr.read\nfw=open('%s','wb')\nw=fw.write" % (src, dest))
627
        while True:
628
            data_len = int(self.exec_("d=r(%u)\nw(d)\nprint(len(d))" % chunk_size))
629
            if not data_len:
630
                break
631
            if progress_callback:
632
                written += data_len
633
                progress_callback(written, src_size)
634
        self.exec_("fr.close()\nfw.close()")
635

636
    def fs_get(self, src, dest, chunk_size=256, progress_callback=None):
637
        if progress_callback:
638
            src_size = self.fs_stat(src).st_size
639
            written = 0
640
        self.exec_("f=open('%s','rb')\nr=f.read" % src)
641
        with open(dest, "wb") as f:
642
            while True:
643
                data = bytearray()
644
                self.exec_("print(r(%u))" % chunk_size, data_consumer=lambda d: data.extend(d))
645
                assert data.endswith(b"\r\n\x04")
646
                try:
647
                    data = ast.literal_eval(str(data[:-3], "ascii"))
648
                    if not isinstance(data, bytes):
649
                        raise ValueError("Not bytes")
650
                except (UnicodeError, ValueError) as e:
651
                    raise PyboardError("fs_get: Could not interpret received data: %s" % str(e))
652
                if not data:
653
                    break
654
                f.write(data)
655
                if progress_callback:
656
                    written += len(data)
657
                    progress_callback(written, src_size)
658
        self.exec_("f.close()")
659

660
    def fs_put(self, src, dest, chunk_size=256, progress_callback=None):
661
        if progress_callback:
662
            src_size = os.path.getsize(src)
663
            written = 0
664
        self.exec_("f=open('%s','wb')\nw=f.write" % dest)
665
        with open(src, "rb") as f:
666
            while True:
667
                data = f.read(chunk_size)
668
                if not data:
669
                    break
670
                if sys.version_info < (3,):
671
                    self.exec_("w(b" + repr(data) + ")")
672
                else:
673
                    self.exec_("w(" + repr(data) + ")")
674
                if progress_callback:
675
                    written += len(data)
676
                    progress_callback(written, src_size)
677
        self.exec_("f.close()")
678

679
    def fs_mkdir(self, dir):
680
        self.exec_("import os\nos.mkdir('%s')" % dir)
681

682
    def fs_rmdir(self, dir):
683
        self.exec_("import os\nos.rmdir('%s')" % dir)
684

685
    def fs_rm(self, src):
686
        self.exec_("import os\nos.remove('%s')" % src)
687

688
    def fs_touch(self, src):
689
        self.exec_("f=open('%s','a')\nf.close()" % src)
690

691

692
# in Python2 exec is a keyword so one must use "exec_"
693
# but for Python3 we want to provide the nicer version "exec"
694
setattr(Pyboard, "exec", Pyboard.exec_)
695

696

697
def execfile(filename, device="/dev/ttyACM0", baudrate=115200, user="micro", password="python"):
698
    pyb = Pyboard(device, baudrate, user, password)
699
    pyb.enter_raw_repl()
700
    output = pyb.execfile(filename)
701
    stdout_write_bytes(output)
702
    pyb.exit_raw_repl()
703
    pyb.close()
704

705

706
def filesystem_command(pyb, args, progress_callback=None, verbose=False):
707
    def fname_remote(src):
708
        if src.startswith(":"):
709
            src = src[1:]
710
        # Convert all path separators to "/", because that's what a remote device uses.
711
        return src.replace(os.path.sep, "/")
712

713
    def fname_cp_dest(src, dest):
714
        _, src = os.path.split(src)
715
        if dest is None or dest == "":
716
            dest = src
717
        elif dest == ".":
718
            dest = "./" + src
719
        elif dest.endswith("/"):
720
            dest += src
721
        return dest
722

723
    cmd = args[0]
724
    args = args[1:]
725
    try:
726
        if cmd == "cp":
727
            if len(args) == 1:
728
                raise PyboardError(
729
                    "cp: missing destination file operand after '{}'".format(args[0])
730
                )
731
            srcs = args[:-1]
732
            dest = args[-1]
733
            if dest.startswith(":"):
734
                op_remote_src = pyb.fs_cp
735
                op_local_src = pyb.fs_put
736
            else:
737
                op_remote_src = pyb.fs_get
738
                op_local_src = lambda src, dest, **_: __import__("shutil").copy(src, dest)
739
            for src in srcs:
740
                if verbose:
741
                    print("cp %s %s" % (src, dest))
742
                if src.startswith(":"):
743
                    op = op_remote_src
744
                else:
745
                    op = op_local_src
746
                src2 = fname_remote(src)
747
                dest2 = fname_cp_dest(src2, fname_remote(dest))
748
                op(src2, dest2, progress_callback=progress_callback)
749
        else:
750
            ops = {
751
                "cat": pyb.fs_cat,
752
                "ls": pyb.fs_ls,
753
                "mkdir": pyb.fs_mkdir,
754
                "rm": pyb.fs_rm,
755
                "rmdir": pyb.fs_rmdir,
756
                "touch": pyb.fs_touch,
757
            }
758
            if cmd not in ops:
759
                raise PyboardError("'{}' is not a filesystem command".format(cmd))
760
            if cmd == "ls" and not args:
761
                args = [""]
762
            for src in args:
763
                src = fname_remote(src)
764
                if verbose:
765
                    print("%s :%s" % (cmd, src))
766
                ops[cmd](src)
767
    except PyboardError as er:
768
        if len(er.args) > 1:
769
            print(str(er.args[2], "ascii"))
770
        else:
771
            print(er)
772
        pyb.exit_raw_repl()
773
        pyb.close()
774
        sys.exit(1)
775

776

777
_injected_import_hook_code = """\
778
import os, io
779
class _FS:
780
  class File(io.IOBase):
781
    def __init__(self):
782
      self.off = 0
783
    def ioctl(self, request, arg):
784
      return 0
785
    def readinto(self, buf):
786
      buf[:] = memoryview(_injected_buf)[self.off:self.off + len(buf)]
787
      self.off += len(buf)
788
      return len(buf)
789
  mount = umount = chdir = lambda *args: None
790
  def stat(self, path):
791
    if path == '_injected.mpy':
792
      return tuple(0 for _ in range(10))
793
    else:
794
      raise OSError(-2) # ENOENT
795
  def open(self, path, mode):
796
    return self.File()
797
os.mount(_FS(), '/_')
798
os.chdir('/_')
799
from _injected import *
800
os.umount('/_')
801
del _injected_buf, _FS
802
"""
803

804

805
def main():
806
    import argparse
807

808
    cmd_parser = argparse.ArgumentParser(description="Run scripts on the pyboard.")
809
    cmd_parser.add_argument(
810
        "-d",
811
        "--device",
812
        default=os.environ.get("PYBOARD_DEVICE", "/dev/ttyACM0"),
813
        help="the serial device or the IP address of the pyboard",
814
    )
815
    cmd_parser.add_argument(
816
        "-b",
817
        "--baudrate",
818
        default=os.environ.get("PYBOARD_BAUDRATE", "115200"),
819
        help="the baud rate of the serial device",
820
    )
821
    cmd_parser.add_argument("-u", "--user", default="micro", help="the telnet login username")
822
    cmd_parser.add_argument("-p", "--password", default="python", help="the telnet login password")
823
    cmd_parser.add_argument("-c", "--command", help="program passed in as string")
824
    cmd_parser.add_argument(
825
        "-w",
826
        "--wait",
827
        default=0,
828
        type=int,
829
        help="seconds to wait for USB connected board to become available",
830
    )
831
    group = cmd_parser.add_mutually_exclusive_group()
832
    group.add_argument(
833
        "--soft-reset",
834
        default=True,
835
        action="store_true",
836
        help="Whether to perform a soft reset when connecting to the board [default]",
837
    )
838
    group.add_argument(
839
        "--no-soft-reset",
840
        action="store_false",
841
        dest="soft_reset",
842
    )
843
    group = cmd_parser.add_mutually_exclusive_group()
844
    group.add_argument(
845
        "--follow",
846
        action="store_true",
847
        default=None,
848
        help="follow the output after running the scripts [default if no scripts given]",
849
    )
850
    group.add_argument(
851
        "--no-follow",
852
        action="store_false",
853
        dest="follow",
854
    )
855
    group = cmd_parser.add_mutually_exclusive_group()
856
    group.add_argument(
857
        "--exclusive",
858
        action="store_true",
859
        default=True,
860
        help="Open the serial device for exclusive access [default]",
861
    )
862
    group.add_argument(
863
        "--no-exclusive",
864
        action="store_false",
865
        dest="exclusive",
866
    )
867
    cmd_parser.add_argument(
868
        "-f",
869
        "--filesystem",
870
        action="store_true",
871
        help="perform a filesystem action: "
872
        "cp local :device | cp :device local | cat path | ls [path] | rm path | mkdir path | rmdir path",
873
    )
874
    cmd_parser.add_argument("files", nargs="*", help="input files")
875
    args = cmd_parser.parse_args()
876

877
    # open the connection to the pyboard
878
    try:
879
        pyb = Pyboard(
880
            args.device, args.baudrate, args.user, args.password, args.wait, args.exclusive
881
        )
882
    except PyboardError as er:
883
        print(er)
884
        sys.exit(1)
885

886
    # run any command or file(s)
887
    if args.command is not None or args.filesystem or len(args.files):
888
        # we must enter raw-REPL mode to execute commands
889
        # this will do a soft-reset of the board
890
        try:
891
            pyb.enter_raw_repl(args.soft_reset)
892
        except PyboardError as er:
893
            print(er)
894
            pyb.close()
895
            sys.exit(1)
896

897
        def execbuffer(buf):
898
            try:
899
                if args.follow is None or args.follow:
900
                    ret, ret_err = pyb.exec_raw(
901
                        buf, timeout=None, data_consumer=stdout_write_bytes
902
                    )
903
                else:
904
                    pyb.exec_raw_no_follow(buf)
905
                    ret_err = None
906
            except PyboardError as er:
907
                print(er)
908
                pyb.close()
909
                sys.exit(1)
910
            except KeyboardInterrupt:
911
                sys.exit(1)
912
            if ret_err:
913
                pyb.exit_raw_repl()
914
                pyb.close()
915
                stdout_write_bytes(ret_err)
916
                sys.exit(1)
917

918
        # do filesystem commands, if given
919
        if args.filesystem:
920
            filesystem_command(pyb, args.files, verbose=True)
921
            del args.files[:]
922

923
        # run the command, if given
924
        if args.command is not None:
925
            execbuffer(args.command.encode("utf-8"))
926

927
        # run any files
928
        for filename in args.files:
929
            with open(filename, "rb") as f:
930
                pyfile = f.read()
931
                if filename.endswith(".mpy") and pyfile[0] == ord("M"):
932
                    pyb.exec_("_injected_buf=" + repr(pyfile))
933
                    pyfile = _injected_import_hook_code
934
                execbuffer(pyfile)
935

936
        # exiting raw-REPL just drops to friendly-REPL mode
937
        pyb.exit_raw_repl()
938

939
    # if asked explicitly, or no files given, then follow the output
940
    if args.follow or (args.command is None and not args.filesystem and len(args.files) == 0):
941
        try:
942
            ret, ret_err = pyb.follow(timeout=None, data_consumer=stdout_write_bytes)
943
        except PyboardError as er:
944
            print(er)
945
            sys.exit(1)
946
        except KeyboardInterrupt:
947
            sys.exit(1)
948
        if ret_err:
949
            pyb.close()
950
            stdout_write_bytes(ret_err)
951
            sys.exit(1)
952

953
    # close the connection to the pyboard
954
    pyb.close()
955

956

957
if __name__ == "__main__":
958
    main()