mirrors/micropython
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

tools/mpremote: Improve error output.

Browse Source 
Makes the filesystem command give standard error messages rather than
just printing the exception from the device.

Makes the distinction between CommandError and TransportError clearer.

This work was funded through GitHub Sponsors.

Signed-off-by: Jim Mussared <jim.mussared@gmail.com>
This commit is contained in:
Jim Mussared 2023-06-15 18:09:39 +10:00 committed by Damien George
parent 6f8157d880
commit dd6f78f014
3 changed files with 70 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tools/mpremote/mpremote/commands.py
View File

@ -63,8 +63,7 @@ def do_connect(state, args=None):
msg = er.args[0]
if msg.startswith("failed to access"):
msg += " (it may be in use by another program)"
print(msg)
sys.exit(1)
raise CommandError(msg)
def do_disconnect(state, _args=None):
@ -322,6 +321,7 @@ def do_filesystem(state, args):
if command == "ls" and not paths:
paths = [""]
try:
# Handle each path sequentially.
for path in paths:
if verbose:
@ -355,6 +355,14 @@ def do_filesystem(state, args):
do_filesystem_recursive_cp(state, path, cp_dest, len(paths) > 1)
else:
do_filesystem_cp(state, path, cp_dest, len(paths) > 1)
except FileNotFoundError as er:
raise CommandError("{}: {}: No such file or directory.".format(command, er.args[0]))
except IsADirectoryError as er:
raise CommandError("{}: {}: Is a directory.".format(command, er.args[0]))
except FileExistsError as er:
raise CommandError("{}: {}: File exists.".format(command, er.args[0]))
except TransportError as er:
raise CommandError("Error with transport:\n{}".format(er.args[0]))
def do_edit(state, args):
@ -362,7 +370,7 @@ def do_edit(state, args):
state.did_action()
if not os.getenv("EDITOR"):
raise TransportError("edit: $EDITOR not set")
raise CommandError("edit: $EDITOR not set")
for src in args.files:
src = src.lstrip(":")
dest_fd, dest = tempfile.mkstemp(suffix=os.path.basename(src))
@ -393,8 +401,7 @@ def _do_execbuffer(state, buf, follow):
stdout_write_bytes(ret_err)
sys.exit(1)
except TransportError as er:
print(er)
sys.exit(1)
raise CommandError(er.args[0])
except KeyboardInterrupt:
sys.exit(1)

36
tools/mpremote/mpremote/transport.py
View File

@ -42,19 +42,27 @@ class TransportError(Exception):
pass
class TransportExecError(TransportError):
def __init__(self, status_code, error_output):
self.status_code = status_code
self.error_output = error_output
super().__init__(error_output)
listdir_result = namedtuple("dir_result", ["name", "st_mode", "st_ino", "st_size"])
# Takes a Transport error (containing the text of an OSError traceback) and
# raises it as the corresponding OSError-derived exception.
def _convert_filesystem_error(e, info):
if len(e.args) >= 3:
if b"OSError" in e.args[2] and b"ENOENT" in e.args[2]:
if "OSError" in e.error_output and "ENOENT" in e.error_output:
return FileNotFoundError(info)
if b"OSError" in e.args[2] and b"EISDIR" in e.args[2]:
if "OSError" in e.error_output and "EISDIR" in e.error_output:
return IsADirectoryError(info)
if b"OSError" in e.args[2] and b"EEXIST" in e.args[2]:
if "OSError" in e.error_output and "EEXIST" in e.error_output:
return FileExistsError(info)
if "OSError" in e.error_output and "ENODEV" in e.error_output:
return FileNotFoundError(info)
return e
@ -72,7 +80,7 @@ class Transport:
buf.extend(b"[")
self.exec(cmd, data_consumer=repr_consumer)
buf.extend(b"]")
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
return [
@ -84,7 +92,7 @@ class Transport:
try:
self.exec("import os")
return os.stat_result(self.eval("os.stat(%s)" % ("'%s'" % src)))
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
def fs_exists(self, src):
@ -109,7 +117,7 @@ class Transport:
)
try:
self.exec(cmd, data_consumer=stdout_write_bytes)
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
def fs_readfile(self, src, chunk_size=256, progress_callback=None):
@ -128,7 +136,7 @@ class Transport:
if progress_callback:
progress_callback(len(contents), src_size)
self.exec("f.close()")
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, src) from None
return contents
@ -148,37 +156,37 @@ class Transport:
if progress_callback:
progress_callback(written, src_size)
self.exec("f.close()")
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, dest) from None
def fs_mkdir(self, path):
try:
self.exec("import os\nos.mkdir('%s')" % path)
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_rmdir(self, path):
try:
self.exec("import os\nos.rmdir('%s')" % path)
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_rmfile(self, path):
try:
self.exec("import os\nos.remove('%s')" % path)
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_touchfile(self, path):
try:
self.exec("f=open('%s','a')\nf.close()" % path)
except TransportError as e:
except TransportExecError as e:
raise _convert_filesystem_error(e, path) from None
def fs_hashfile(self, path, algo, chunk_size=256):
try:
self.exec("import hashlib\nh = hashlib.{algo}()".format(algo=algo))
except TransportError:
except TransportExecError:
# hashlib (or hashlib.{algo}) not available on device. Do the hash locally.
data = self.fs_readfile(path, chunk_size=chunk_size)
return getattr(hashlib, algo)(data).digest()

4
tools/mpremote/mpremote/transport_serial.py
View File

@ -38,7 +38,7 @@
import ast, io, os, re, struct, sys, time
from errno import EPERM
from .console import VT_ENABLED
from .transport import TransportError, Transport
from .transport import TransportError, TransportExecError, Transport
class SerialTransport(Transport):
@ -267,7 +267,7 @@ class SerialTransport(Transport):
def exec(self, command, data_consumer=None):
ret, ret_err = self.exec_raw(command, data_consumer=data_consumer)
if ret_err:
raise TransportError("exception", ret, ret_err)
raise TransportExecError(ret, ret_err.decode())
return ret
def execfile(self, filename):