mirror of
https://github.com/DarkFlippers/unleashed-firmware.git
synced 2024-12-22 21:01:40 +03:00
387 lines
12 KiB
Python
387 lines
12 KiB
Python
import os
|
|
import sys
|
|
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, chunk_size: int = 8192):
|
|
self.port = serial.Serial()
|
|
self.port.port = portname
|
|
self.port.timeout = 2
|
|
self.port.baudrate = 115200 # Doesn't matter for VCP
|
|
self.read = BufferedRead(self.port)
|
|
self.last_error = ""
|
|
self.chunk_size = chunk_size
|
|
|
|
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(f'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)
|
|
|
|
with open(filename_from, "rb") as file:
|
|
filesize = os.fstat(file.fileno()).st_size
|
|
|
|
buffer_size = self.chunk_size
|
|
while True:
|
|
filedata = file.read(buffer_size)
|
|
size = len(filedata)
|
|
if size == 0:
|
|
break
|
|
|
|
self.send_and_wait_eol(f'storage write_chunk "{filename_to}" {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)
|
|
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))
|
|
sys.stdout.write(
|
|
f"\r{percent}%, chunk {current_chunk} of {total_chunks}"
|
|
)
|
|
sys.stdout.flush()
|
|
print()
|
|
return True
|
|
|
|
def read_file(self, filename):
|
|
"""Receive file from Flipper, and get filedata (bytes)"""
|
|
buffer_size = self.chunk_size
|
|
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])
|
|
read_size = 0
|
|
|
|
while read_size < size:
|
|
self.read.until("Ready?" + self.CLI_EOL)
|
|
self.send("y")
|
|
chunk_size = min(size - read_size, buffer_size)
|
|
filedata.extend(self.port.read(chunk_size))
|
|
read_size = read_size + chunk_size
|
|
|
|
percent = str(math.ceil(read_size / size * 100))
|
|
total_chunks = str(math.ceil(size / buffer_size))
|
|
current_chunk = str(math.ceil(read_size / buffer_size))
|
|
sys.stdout.write(f"\r{percent}%, chunk {current_chunk} of {total_chunks}")
|
|
sys.stdout.flush()
|
|
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 format_ext(self):
|
|
"""Create a directory on Flipper"""
|
|
self.send_and_wait_eol("storage format /ext\r")
|
|
self.send_and_wait_eol("y\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(self.chunk_size), 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")
|