[FL-1682] Faster file receiving function. Storage management scripts. (#647)
* Storage-cli: the file write function is splitted into a function for working with text and function for raw data * Storage-cli: read_chunks, renamed write_raw to write_chunk * Storage-cli: fix typo * SD Hal: fixed wrong read/write block address * HAL-console: printf * Storage benchmark: more popular sizes * Toolbox: md5 * Storage-cli: better read_chunks function, md5 hash function * Notification: fixed incorrect error message when loading settings * Storage-cli: stat command * Storage-cli: stat, "/" is also storage * Scripts: add storage managment script * Scripts, storage lib: send command with known response syntax instead of SOH * Scripts: python3 from env * Storage-cli: fixed mess with error texts * Storage-cli: write, show welcome message only if we didn't have any errors * Storage: poorly fixed folders copying * Storage: close an old file if an error occurred while opening a new file * Storage-cli: fixed storage info spacing * Scripts: use positional arguments in storage. * Scripts: explicit string encoding and decoding, documentation in comments. Co-authored-by: あく <alleteam@gmail.com>
This commit is contained in:
SG
GitHub
0
scripts/flipper/__init__.py
Normal file
0
scripts/flipper/__init__.py
Normal file
375
scripts/flipper/storage.py
Normal file
375
scripts/flipper/storage.py
Normal file
@@ -0,0 +1,375 @@
|
||||
import os
|
||||
import serial
|
||||
import time
|
||||
import hashlib
|
||||
import math
|
||||
|
||||
|
||||
def timing(func):
|
||||
"""
|
||||
Speedometer decorator
|
||||
"""
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
time1 = time.monotonic()
|
||||
ret = func(*args, **kwargs)
|
||||
time2 = time.monotonic()
|
||||
print(
|
||||
"{:s} function took {:.3f} ms".format(
|
||||
func.__name__, (time2 - time1) * 1000.0
|
||||
)
|
||||
)
|
||||
return ret
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
class BufferedRead:
|
||||
def __init__(self, stream):
|
||||
self.buffer = bytearray()
|
||||
self.stream = stream
|
||||
|
||||
def until(self, eol="\n", cut_eol=True):
|
||||
eol = eol.encode("ascii")
|
||||
while True:
|
||||
# search in buffer
|
||||
i = self.buffer.find(eol)
|
||||
if i >= 0:
|
||||
if cut_eol:
|
||||
read = self.buffer[:i]
|
||||
else:
|
||||
read = self.buffer[: i + len(eol)]
|
||||
self.buffer = self.buffer[i + len(eol) :]
|
||||
return read
|
||||
|
||||
# read and append to buffer
|
||||
i = max(1, self.stream.in_waiting)
|
||||
data = self.stream.read(i)
|
||||
self.buffer.extend(data)
|
||||
|
||||
|
||||
class FlipperStorage:
|
||||
CLI_PROMPT = ">: "
|
||||
CLI_EOL = "\r\n"
|
||||
|
||||
def __init__(self, portname: str):
|
||||
self.port = serial.Serial()
|
||||
self.port.port = portname
|
||||
self.port.timeout = 2
|
||||
self.port.baudrate = 115200
|
||||
self.read = BufferedRead(self.port)
|
||||
self.last_error = ""
|
||||
|
||||
def start(self):
|
||||
self.port.open()
|
||||
self.port.reset_input_buffer()
|
||||
# Send a command with a known syntax to make sure the buffer is flushed
|
||||
self.send("device_info\r")
|
||||
self.read.until("hardware_model :")
|
||||
# And read buffer until we get prompt
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
def stop(self):
|
||||
self.port.close()
|
||||
|
||||
def send(self, line):
|
||||
self.port.write(line.encode("ascii"))
|
||||
|
||||
def send_and_wait_eol(self, line):
|
||||
self.send(line)
|
||||
return self.read.until(self.CLI_EOL)
|
||||
|
||||
def send_and_wait_prompt(self, line):
|
||||
self.send(line)
|
||||
return self.read.until(self.CLI_PROMPT)
|
||||
|
||||
def has_error(self, data):
|
||||
"""Is data has error"""
|
||||
if data.find(b"Storage error") != -1:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def get_error(self, data):
|
||||
"""Extract error text from data and print it"""
|
||||
error, error_text = data.decode("ascii").split(": ")
|
||||
return error_text.strip()
|
||||
|
||||
def list_tree(self, path="/", level=0):
|
||||
"""List files and dirs on Flipper"""
|
||||
path = path.replace("//", "/")
|
||||
|
||||
self.send_and_wait_eol('storage list "' + path + '"\r')
|
||||
|
||||
data = self.read.until(self.CLI_PROMPT)
|
||||
lines = data.split(b"\r\n")
|
||||
|
||||
for line in lines:
|
||||
try:
|
||||
# TODO: better decoding, considering non-ascii characters
|
||||
line = line.decode("ascii")
|
||||
except:
|
||||
continue
|
||||
|
||||
line = line.strip()
|
||||
|
||||
if len(line) == 0:
|
||||
continue
|
||||
|
||||
if self.has_error(line.encode("ascii")):
|
||||
print(self.get_error(line.encode("ascii")))
|
||||
continue
|
||||
|
||||
if line == "Empty":
|
||||
continue
|
||||
|
||||
type, info = line.split(" ", 1)
|
||||
if type == "[D]":
|
||||
# Print directory name
|
||||
print((path + "/" + info).replace("//", "/"))
|
||||
# And recursively go inside
|
||||
self.list_tree(path + "/" + info, level + 1)
|
||||
elif type == "[F]":
|
||||
name, size = info.rsplit(" ", 1)
|
||||
# Print file name and size
|
||||
print((path + "/" + name).replace("//", "/") + ", size " + size)
|
||||
else:
|
||||
# Something wrong, pass
|
||||
pass
|
||||
|
||||
def walk(self, path="/"):
|
||||
dirs = []
|
||||
nondirs = []
|
||||
walk_dirs = []
|
||||
|
||||
path = path.replace("//", "/")
|
||||
self.send_and_wait_eol('storage list "' + path + '"\r')
|
||||
data = self.read.until(self.CLI_PROMPT)
|
||||
lines = data.split(b"\r\n")
|
||||
|
||||
for line in lines:
|
||||
try:
|
||||
# TODO: better decoding, considering non-ascii characters
|
||||
line = line.decode("ascii")
|
||||
except:
|
||||
continue
|
||||
|
||||
line = line.strip()
|
||||
|
||||
if len(line) == 0:
|
||||
continue
|
||||
|
||||
if self.has_error(line.encode("ascii")):
|
||||
continue
|
||||
|
||||
if line == "Empty":
|
||||
continue
|
||||
|
||||
type, info = line.split(" ", 1)
|
||||
if type == "[D]":
|
||||
# Print directory name
|
||||
dirs.append(info)
|
||||
walk_dirs.append((path + "/" + info).replace("//", "/"))
|
||||
|
||||
elif type == "[F]":
|
||||
name, size = info.rsplit(" ", 1)
|
||||
# Print file name and size
|
||||
nondirs.append(name)
|
||||
else:
|
||||
# Something wrong, pass
|
||||
pass
|
||||
|
||||
# topdown walk, yield before recursy
|
||||
yield path, dirs, nondirs
|
||||
for new_path in walk_dirs:
|
||||
yield from self.walk(new_path)
|
||||
|
||||
def send_file(self, filename_from, filename_to):
|
||||
"""Send file from local device to Flipper"""
|
||||
self.remove(filename_to)
|
||||
|
||||
file = open(filename_from, "rb")
|
||||
filesize = os.fstat(file.fileno()).st_size
|
||||
|
||||
buffer_size = 512
|
||||
while True:
|
||||
filedata = file.read(buffer_size)
|
||||
size = len(filedata)
|
||||
if size == 0:
|
||||
break
|
||||
|
||||
self.send_and_wait_eol(
|
||||
'storage write_chunk "' + filename_to + '" ' + str(size) + "\r"
|
||||
)
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
file.close()
|
||||
return False
|
||||
|
||||
self.port.write(filedata)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
percent = str(math.ceil(file.tell() / filesize * 100))
|
||||
total_chunks = str(math.ceil(filesize / buffer_size))
|
||||
current_chunk = str(math.ceil(file.tell() / buffer_size))
|
||||
print(
|
||||
percent + "%, chunk " + current_chunk + " of " + total_chunks, end="\r"
|
||||
)
|
||||
file.close()
|
||||
print()
|
||||
return True
|
||||
|
||||
def read_file(self, filename):
|
||||
"""Receive file from Flipper, and get filedata (bytes)"""
|
||||
buffer_size = 512
|
||||
self.send_and_wait_eol(
|
||||
'storage read_chunks "' + filename + '" ' + str(buffer_size) + "\r"
|
||||
)
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
filedata = bytearray()
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
return filedata
|
||||
size = int(answer.split(b": ")[1])
|
||||
readed_size = 0
|
||||
|
||||
while readed_size < size:
|
||||
self.read.until("Ready?" + self.CLI_EOL)
|
||||
self.send("y")
|
||||
read_size = min(size - readed_size, buffer_size)
|
||||
filedata.extend(self.port.read(read_size))
|
||||
readed_size = readed_size + read_size
|
||||
|
||||
percent = str(math.ceil(readed_size / size * 100))
|
||||
total_chunks = str(math.ceil(size / buffer_size))
|
||||
current_chunk = str(math.ceil(readed_size / buffer_size))
|
||||
print(
|
||||
percent + "%, chunk " + current_chunk + " of " + total_chunks, end="\r"
|
||||
)
|
||||
print()
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
return filedata
|
||||
|
||||
def receive_file(self, filename_from, filename_to):
|
||||
"""Receive file from Flipper to local storage"""
|
||||
with open(filename_to, "wb") as file:
|
||||
data = self.read_file(filename_from)
|
||||
if not data:
|
||||
return False
|
||||
else:
|
||||
file.write(data)
|
||||
return True
|
||||
|
||||
def exist(self, path):
|
||||
"""Is file or dir exist on Flipper"""
|
||||
self.send_and_wait_eol('storage stat "' + path + '"\r')
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def exist_dir(self, path):
|
||||
"""Is dir exist on Flipper"""
|
||||
self.send_and_wait_eol('storage stat "' + path + '"\r')
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
return False
|
||||
else:
|
||||
if answer.find(b"Directory") != -1:
|
||||
return True
|
||||
elif answer.find(b"Storage") != -1:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def exist_file(self, path):
|
||||
"""Is file exist on Flipper"""
|
||||
self.send_and_wait_eol('storage stat "' + path + '"\r')
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
return False
|
||||
else:
|
||||
if answer.find(b"File, size:") != -1:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def size(self, path):
|
||||
"""file size on Flipper"""
|
||||
self.send_and_wait_eol('storage stat "' + path + '"\r')
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
return False
|
||||
else:
|
||||
if answer.find(b"File, size:") != -1:
|
||||
size = int(
|
||||
"".join(
|
||||
ch
|
||||
for ch in answer.split(b": ")[1].decode("ascii")
|
||||
if ch.isdigit()
|
||||
)
|
||||
)
|
||||
return size
|
||||
else:
|
||||
self.last_error = "access denied"
|
||||
return -1
|
||||
|
||||
def mkdir(self, path):
|
||||
"""Create a directory on Flipper"""
|
||||
self.send_and_wait_eol('storage mkdir "' + path + '"\r')
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def remove(self, path):
|
||||
"""Remove file or directory on Flipper"""
|
||||
self.send_and_wait_eol('storage remove "' + path + '"\r')
|
||||
answer = self.read.until(self.CLI_EOL)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
if self.has_error(answer):
|
||||
self.last_error = self.get_error(answer)
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def hash_local(self, filename):
|
||||
"""Hash of local file"""
|
||||
hash_md5 = hashlib.md5()
|
||||
with open(filename, "rb") as f:
|
||||
for chunk in iter(lambda: f.read(4096), b""):
|
||||
hash_md5.update(chunk)
|
||||
return hash_md5.hexdigest()
|
||||
|
||||
def hash_flipper(self, filename):
|
||||
"""Get hash of file on Flipper"""
|
||||
self.send_and_wait_eol('storage md5 "' + filename + '"\r')
|
||||
hash = self.read.until(self.CLI_EOL)
|
||||
self.read.until(self.CLI_PROMPT)
|
||||
|
||||
if self.has_error(hash):
|
||||
self.last_error = self.get_error(hash)
|
||||
return ""
|
||||
else:
|
||||
return hash.decode("ascii")
|
||||
267
scripts/storage.py
Executable file
267
scripts/storage.py
Executable file
@@ -0,0 +1,267 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from flipper.storage import FlipperStorage
|
||||
import logging
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
import binascii
|
||||
import posixpath
|
||||
|
||||
|
||||
class Main:
|
||||
def __init__(self):
|
||||
# command args
|
||||
self.parser = argparse.ArgumentParser()
|
||||
self.parser.add_argument("-d", "--debug", action="store_true", help="Debug")
|
||||
self.parser.add_argument("-p", "--port", help="CDC Port", required=True)
|
||||
self.subparsers = self.parser.add_subparsers(help="sub-command help")
|
||||
|
||||
self.parser_mkdir = self.subparsers.add_parser("mkdir", help="Create directory")
|
||||
self.parser_mkdir.add_argument("flipper_path", help="Flipper path")
|
||||
self.parser_mkdir.set_defaults(func=self.mkdir)
|
||||
|
||||
self.parser_remove = self.subparsers.add_parser(
|
||||
"remove", help="Remove file/directory"
|
||||
)
|
||||
self.parser_remove.add_argument("flipper_path", help="Flipper path")
|
||||
self.parser_remove.set_defaults(func=self.remove)
|
||||
|
||||
self.parser_read = self.subparsers.add_parser("read", help="Read file")
|
||||
self.parser_read.add_argument("flipper_path", help="Flipper path")
|
||||
self.parser_read.set_defaults(func=self.read)
|
||||
|
||||
self.parser_size = self.subparsers.add_parser("size", help="Size of file")
|
||||
self.parser_size.add_argument("flipper_path", help="Flipper path")
|
||||
self.parser_size.set_defaults(func=self.size)
|
||||
|
||||
self.parser_receive = self.subparsers.add_parser("receive", help="Receive file")
|
||||
self.parser_receive.add_argument("flipper_path", help="Flipper path")
|
||||
self.parser_receive.add_argument("local_path", help="Local path")
|
||||
self.parser_receive.set_defaults(func=self.receive)
|
||||
|
||||
self.parser_send = self.subparsers.add_parser(
|
||||
"send", help="Send file or directory"
|
||||
)
|
||||
self.parser_send.add_argument(
|
||||
"-f", "--force", help="Force sending", action="store_true"
|
||||
)
|
||||
self.parser_send.add_argument("local_path", help="Local path")
|
||||
self.parser_send.add_argument("flipper_path", help="Flipper path")
|
||||
self.parser_send.set_defaults(func=self.send)
|
||||
|
||||
self.parser_list = self.subparsers.add_parser(
|
||||
"list", help="Recursively list files and dirs"
|
||||
)
|
||||
self.parser_list.add_argument("flipper_path", help="Flipper path", default="/")
|
||||
self.parser_list.set_defaults(func=self.list)
|
||||
|
||||
# logging
|
||||
self.logger = logging.getLogger()
|
||||
|
||||
def __call__(self):
|
||||
self.args = self.parser.parse_args()
|
||||
if "func" not in self.args:
|
||||
self.parser.error("Choose something to do")
|
||||
# configure log output
|
||||
self.log_level = logging.DEBUG if self.args.debug else logging.INFO
|
||||
self.logger.setLevel(self.log_level)
|
||||
self.handler = logging.StreamHandler(sys.stdout)
|
||||
self.handler.setLevel(self.log_level)
|
||||
self.formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
|
||||
self.handler.setFormatter(self.formatter)
|
||||
self.logger.addHandler(self.handler)
|
||||
# execute requested function
|
||||
self.args.func()
|
||||
|
||||
def mkdir(self):
|
||||
storage = FlipperStorage(self.args.port)
|
||||
storage.start()
|
||||
self.logger.debug(f'Creating "{self.args.flipper_path}"')
|
||||
if not storage.mkdir(self.args.flipper_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
storage.stop()
|
||||
|
||||
def remove(self):
|
||||
storage = FlipperStorage(self.args.port)
|
||||
storage.start()
|
||||
self.logger.debug(f'Removing "{self.args.flipper_path}"')
|
||||
if not storage.remove(self.args.flipper_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
storage.stop()
|
||||
|
||||
def receive(self):
|
||||
storage = FlipperStorage(self.args.port)
|
||||
storage.start()
|
||||
|
||||
if storage.exist_dir(self.args.flipper_path):
|
||||
for dirpath, dirnames, filenames in storage.walk(self.args.flipper_path):
|
||||
self.logger.debug(
|
||||
f'Processing directory "{os.path.normpath(dirpath)}"'.replace(
|
||||
os.sep, "/"
|
||||
)
|
||||
)
|
||||
dirnames.sort()
|
||||
filenames.sort()
|
||||
|
||||
rel_path = os.path.relpath(dirpath, self.args.flipper_path)
|
||||
|
||||
for dirname in dirnames:
|
||||
local_dir_path = os.path.join(
|
||||
self.args.local_path, rel_path, dirname
|
||||
)
|
||||
local_dir_path = os.path.normpath(local_dir_path)
|
||||
os.makedirs(local_dir_path, exist_ok=True)
|
||||
|
||||
for filename in filenames:
|
||||
local_file_path = os.path.join(
|
||||
self.args.local_path, rel_path, filename
|
||||
)
|
||||
local_file_path = os.path.normpath(local_file_path)
|
||||
flipper_file_path = os.path.normpath(
|
||||
os.path.join(dirpath, filename)
|
||||
).replace(os.sep, "/")
|
||||
self.logger.info(
|
||||
f'Receiving "{flipper_file_path}" to "{local_file_path}"'
|
||||
)
|
||||
if not storage.receive_file(flipper_file_path, local_file_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
|
||||
else:
|
||||
self.logger.info(
|
||||
f'Receiving "{self.args.flipper_path}" to "{self.args.local_path}"'
|
||||
)
|
||||
if not storage.receive_file(self.args.flipper_path, self.args.local_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
storage.stop()
|
||||
|
||||
def send(self):
|
||||
storage = FlipperStorage(self.args.port)
|
||||
storage.start()
|
||||
self.send_to_storage(
|
||||
storage, self.args.flipper_path, self.args.local_path, self.args.force
|
||||
)
|
||||
storage.stop()
|
||||
|
||||
# send file or folder recursively
|
||||
def send_to_storage(self, storage, flipper_path, local_path, force):
|
||||
if not os.path.exists(local_path):
|
||||
self.logger.error(f'Error: "{local_path}" is not exist')
|
||||
|
||||
if os.path.isdir(local_path):
|
||||
# create parent dir
|
||||
self.mkdir_on_storage(storage, flipper_path)
|
||||
|
||||
for dirpath, dirnames, filenames in os.walk(local_path):
|
||||
self.logger.debug(f'Processing directory "{os.path.normpath(dirpath)}"')
|
||||
dirnames.sort()
|
||||
filenames.sort()
|
||||
rel_path = os.path.relpath(dirpath, local_path)
|
||||
|
||||
# create subdirs
|
||||
for dirname in dirnames:
|
||||
flipper_dir_path = os.path.join(flipper_path, rel_path, dirname)
|
||||
flipper_dir_path = os.path.normpath(flipper_dir_path).replace(
|
||||
os.sep, "/"
|
||||
)
|
||||
self.mkdir_on_storage(storage, flipper_dir_path)
|
||||
|
||||
# send files
|
||||
for filename in filenames:
|
||||
flipper_file_path = os.path.join(flipper_path, rel_path, filename)
|
||||
flipper_file_path = os.path.normpath(flipper_file_path).replace(
|
||||
os.sep, "/"
|
||||
)
|
||||
local_file_path = os.path.normpath(os.path.join(dirpath, filename))
|
||||
self.send_file_to_storage(
|
||||
storage, flipper_file_path, local_file_path, force
|
||||
)
|
||||
else:
|
||||
self.send_file_to_storage(storage, flipper_path, local_path, force)
|
||||
|
||||
# make directory with exist check
|
||||
def mkdir_on_storage(self, storage, flipper_dir_path):
|
||||
if not storage.exist_dir(flipper_dir_path):
|
||||
self.logger.debug(f'"{flipper_dir_path}" not exist, creating')
|
||||
if not storage.mkdir(flipper_dir_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
else:
|
||||
self.logger.debug(f'"{flipper_dir_path}" already exist')
|
||||
|
||||
# send file with exist check and hash check
|
||||
def send_file_to_storage(self, storage, flipper_file_path, local_file_path, force):
|
||||
if not storage.exist_file(flipper_file_path):
|
||||
self.logger.debug(
|
||||
f'"{flipper_file_path}" not exist, sending "{local_file_path}"'
|
||||
)
|
||||
self.logger.info(f'Sending "{local_file_path}" to "{flipper_file_path}"')
|
||||
if not storage.send_file(local_file_path, flipper_file_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
elif force:
|
||||
self.logger.debug(
|
||||
f'"{flipper_file_path}" exist, but will be overwritten by "{local_file_path}"'
|
||||
)
|
||||
self.logger.info(f'Sending "{local_file_path}" to "{flipper_file_path}"')
|
||||
if not storage.send_file(local_file_path, flipper_file_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
else:
|
||||
self.logger.debug(
|
||||
f'"{flipper_file_path}" exist, compare hash with "{local_file_path}"'
|
||||
)
|
||||
hash_local = storage.hash_local(local_file_path)
|
||||
hash_flipper = storage.hash_flipper(flipper_file_path)
|
||||
|
||||
if not hash_flipper:
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
|
||||
if hash_local == hash_flipper:
|
||||
self.logger.debug(
|
||||
f'"{flipper_file_path}" are equal to "{local_file_path}"'
|
||||
)
|
||||
else:
|
||||
self.logger.debug(
|
||||
f'"{flipper_file_path}" are not equal to "{local_file_path}"'
|
||||
)
|
||||
self.logger.info(
|
||||
f'Sending "{local_file_path}" to "{flipper_file_path}"'
|
||||
)
|
||||
if not storage.send_file(local_file_path, flipper_file_path):
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
|
||||
def read(self):
|
||||
storage = FlipperStorage(self.args.port)
|
||||
storage.start()
|
||||
self.logger.debug(f'Reading "{self.args.flipper_path}"')
|
||||
data = storage.read_file(self.args.flipper_path)
|
||||
if not data:
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
else:
|
||||
try:
|
||||
print("Text data:")
|
||||
print(data.decode())
|
||||
except:
|
||||
print("Binary hexadecimal data:")
|
||||
print(binascii.hexlify(data).decode())
|
||||
storage.stop()
|
||||
|
||||
def size(self):
|
||||
storage = FlipperStorage(self.args.port)
|
||||
storage.start()
|
||||
self.logger.debug(f'Getting size of "{self.args.flipper_path}"')
|
||||
size = storage.size(self.args.flipper_path)
|
||||
if size < 0:
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
else:
|
||||
print(size)
|
||||
storage.stop()
|
||||
|
||||
def list(self):
|
||||
storage = FlipperStorage(self.args.port)
|
||||
storage.start()
|
||||
self.logger.debug(f'Listing "{self.args.flipper_path}"')
|
||||
storage.list_tree(self.args.flipper_path)
|
||||
storage.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
Main()()
|
||||
Reference in New Issue
Block a user