mirror of
https://github.com/DarkFlippers/unleashed-firmware.git
synced 2025-12-12 20:49:49 +04:00
[FL-3097] fbt, faploader: minimal app module implementation (#2420)
* fbt, faploader: minimal app module implementation * faploader, libs: moved API hashtable core to flipper_application * example: compound api * lib: flipper_application: naming fixes, doxygen comments * fbt: changed `requires` manifest field behavior for app extensions * examples: refactored plugin apps; faploader: changed new API naming; fbt: changed PLUGIN app type meaning * loader: dropped support for debug apps & plugin menus * moved applications/plugins -> applications/external * Restored x bit on chiplist_convert.py * git: fixed free-dap submodule path * pvs: updated submodule paths * examples: example_advanced_plugins.c: removed potential memory leak on errors * examples: example_plugins: refined requires * fbt: not deploying app modules for debug/sample apps; extra validation for .PLUGIN-type apps * apps: removed cdefines for external apps * fbt: moved ext app path definition * fbt: reworked fap_dist handling; f18: synced api_symbols.csv * fbt: removed resources_paths for extapps * scripts: reworked storage * scripts: reworked runfap.py & selfupdate.py to use new api * wip: fal runner * fbt: moved file packaging into separate module * scripts: storage: fixes * scripts: storage: minor fixes for new api * fbt: changed internal artifact storage details for external apps * scripts: storage: additional fixes and better error reporting; examples: using APP_DATA_PATH() * fbt, scripts: reworked launch_app to deploy plugins; moved old runfap.py to distfap.py * fbt: extra check for plugins descriptors * fbt: additional checks in emitter * fbt: better info message on SDK rebuild * scripts: removed requirements.txt * loader: removed remnants of plugins & debug menus * post-review fixes
This commit is contained in:
@@ -1,16 +1,28 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from flipper.app import App
|
||||
from flipper.storage import FlipperStorage
|
||||
from flipper.storage import FlipperStorage, FlipperStorageOperations
|
||||
from flipper.utils.cdc import resolve_port
|
||||
|
||||
import logging
|
||||
import os
|
||||
import binascii
|
||||
import filecmp
|
||||
import tempfile
|
||||
|
||||
|
||||
def WrapStorageOp(func):
|
||||
def wrapper(*args, **kwargs):
|
||||
try:
|
||||
func(*args, **kwargs)
|
||||
return 0
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
# raise # uncomment to debug
|
||||
return 1
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
class Main(App):
|
||||
def init(self):
|
||||
self.parser.add_argument("-p", "--port", help="CDC Port", default="auto")
|
||||
@@ -71,229 +83,71 @@ class Main(App):
|
||||
)
|
||||
self.parser_stress.set_defaults(func=self.stress)
|
||||
|
||||
def _get_storage(self):
|
||||
def _get_port(self):
|
||||
if not (port := resolve_port(self.logger, self.args.port)):
|
||||
return None
|
||||
|
||||
storage = FlipperStorage(port)
|
||||
storage.start()
|
||||
return storage
|
||||
raise Exception("Failed to resolve port")
|
||||
return port
|
||||
|
||||
@WrapStorageOp
|
||||
def mkdir(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
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()
|
||||
return 0
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
storage.mkdir(self.args.flipper_path)
|
||||
|
||||
@WrapStorageOp
|
||||
def remove(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
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()
|
||||
return 0
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
storage.remove(self.args.flipper_path)
|
||||
|
||||
@WrapStorageOp
|
||||
def receive(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
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}"'
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
FlipperStorageOperations(storage).recursive_receive(
|
||||
self.args.flipper_path, 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()
|
||||
return 0
|
||||
|
||||
@WrapStorageOp
|
||||
def send(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
self.send_to_storage(
|
||||
storage, self.args.flipper_path, self.args.local_path, self.args.force
|
||||
)
|
||||
storage.stop()
|
||||
return 0
|
||||
|
||||
# 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}" does 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 exists')
|
||||
|
||||
# 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}" does not exist, sending "{local_file_path}"'
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
FlipperStorageOperations(storage).recursive_send(
|
||||
self.args.flipper_path, self.args.local_path, self.args.force
|
||||
)
|
||||
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}" exists, 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}" exists, 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}" is equal to "{local_file_path}"'
|
||||
)
|
||||
else:
|
||||
self.logger.debug(
|
||||
f'"{flipper_file_path}" is 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}")
|
||||
|
||||
@WrapStorageOp
|
||||
def read(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
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:
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
data = storage.read_file(self.args.flipper_path)
|
||||
try:
|
||||
print("Text data:")
|
||||
print(data.decode())
|
||||
except:
|
||||
print("Binary hexadecimal data:")
|
||||
print(binascii.hexlify(data).decode())
|
||||
storage.stop()
|
||||
return 0
|
||||
|
||||
@WrapStorageOp
|
||||
def size(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
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()
|
||||
return 0
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
print(storage.size(self.args.flipper_path))
|
||||
|
||||
@WrapStorageOp
|
||||
def list(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
self.logger.debug(f'Listing "{self.args.flipper_path}"')
|
||||
storage.list_tree(self.args.flipper_path)
|
||||
storage.stop()
|
||||
return 0
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
storage.list_tree(self.args.flipper_path)
|
||||
|
||||
@WrapStorageOp
|
||||
def format_ext(self):
|
||||
if not (storage := self._get_storage()):
|
||||
return 1
|
||||
|
||||
self.logger.debug("Formatting /ext SD card")
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
storage.format_ext()
|
||||
|
||||
if not storage.format_ext():
|
||||
self.logger.error(f"Error: {storage.last_error}")
|
||||
storage.stop()
|
||||
return 0
|
||||
|
||||
@WrapStorageOp
|
||||
def stress(self):
|
||||
self.logger.error("This test is wearing out flash memory.")
|
||||
self.logger.error("Never use it with internal storage(/int)")
|
||||
self.logger.error("Never use it with internal storage (/int)")
|
||||
|
||||
if self.args.flipper_path.startswith(
|
||||
"/int"
|
||||
@@ -312,24 +166,19 @@ class Main(App):
|
||||
with open(send_file_name, "w") as fout:
|
||||
fout.write("A" * self.args.file_size)
|
||||
|
||||
storage = self._get_storage()
|
||||
if not storage:
|
||||
return 1
|
||||
|
||||
if storage.exist_file(self.args.flipper_path):
|
||||
self.logger.error("File exists, remove it first")
|
||||
return
|
||||
while self.args.count > 0:
|
||||
storage.send_file(send_file_name, self.args.flipper_path)
|
||||
storage.receive_file(self.args.flipper_path, receive_file_name)
|
||||
if not filecmp.cmp(receive_file_name, send_file_name):
|
||||
self.logger.error("Files mismatch")
|
||||
break
|
||||
storage.remove(self.args.flipper_path)
|
||||
os.unlink(receive_file_name)
|
||||
self.args.count -= 1
|
||||
storage.stop()
|
||||
return 0
|
||||
with FlipperStorage(self._get_port()) as storage:
|
||||
if storage.exist_file(self.args.flipper_path):
|
||||
self.logger.error("File exists, remove it first")
|
||||
return
|
||||
while self.args.count > 0:
|
||||
storage.send_file(send_file_name, self.args.flipper_path)
|
||||
storage.receive_file(self.args.flipper_path, receive_file_name)
|
||||
if not filecmp.cmp(receive_file_name, send_file_name):
|
||||
self.logger.error("Files mismatch")
|
||||
break
|
||||
storage.remove(self.args.flipper_path)
|
||||
os.unlink(receive_file_name)
|
||||
self.args.count -= 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user