mirror of
https://github.com/DarkFlippers/unleashed-firmware
synced 2024-11-26 22:40:25 +00:00
b95cd2df14
* Updater fixes related to /int handling updater: performing factory reset on update, checking for LFS free space before updating, fixed improper error handling on backup/restore operations, rebalanced update stage weights for better progress visuals scripts: added CLI output validation for selfupdate.py storage: added pointer validation in storage_int_common_fs_info desktop: fixed crash on rendering invalid slideshows * Typo fix * rpc: Updated protobuf to 0.9 * rpc: removed updater status conversion Co-authored-by: あく <alleteam@gmail.com>
120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from typing import final
|
|
from flipper.app import App
|
|
from flipper.storage import FlipperStorage
|
|
from flipper.utils.cdc import resolve_port
|
|
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import serial.tools.list_ports as list_ports
|
|
|
|
|
|
class Main(App):
|
|
def init(self):
|
|
self.parser.add_argument("-p", "--port", help="CDC Port", default="auto")
|
|
self.parser.add_argument(
|
|
"-b",
|
|
"--baud",
|
|
help="Port Baud rate",
|
|
required=False,
|
|
default=115200 * 4,
|
|
type=int,
|
|
)
|
|
|
|
self.parser.add_argument("manifest_path", help="Manifest path")
|
|
self.parser.add_argument(
|
|
"--pkg_dir_name", help="Update dir name", default="pcbundle", required=False
|
|
)
|
|
self.parser.set_defaults(func=self.install)
|
|
|
|
# logging
|
|
self.logger = logging.getLogger()
|
|
|
|
# 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}")
|
|
return False
|
|
else:
|
|
self.logger.debug(f'"{flipper_dir_path}" already exists')
|
|
return True
|
|
|
|
# send file with exist check and hash check
|
|
def send_file_to_storage(self, storage, flipper_file_path, local_file_path, force):
|
|
exists = storage.exist_file(flipper_file_path)
|
|
do_upload = not exists
|
|
if exists:
|
|
hash_local = storage.hash_local(local_file_path)
|
|
hash_flipper = storage.hash_flipper(flipper_file_path)
|
|
self.logger.debug(f"hash check: local {hash_local}, flipper {hash_flipper}")
|
|
do_upload = force or (hash_local != hash_flipper)
|
|
|
|
if do_upload:
|
|
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}")
|
|
return False
|
|
return True
|
|
|
|
def install(self):
|
|
if not (port := resolve_port(self.logger, self.args.port)):
|
|
return 1
|
|
|
|
storage = FlipperStorage(port, self.args.baud)
|
|
storage.start()
|
|
|
|
try:
|
|
if not os.path.isfile(self.args.manifest_path):
|
|
self.logger.error("Error: manifest not found")
|
|
return 2
|
|
|
|
manifest_path = pathlib.Path(os.path.abspath(self.args.manifest_path))
|
|
manifest_name, pkg_name = manifest_path.parts[-1], manifest_path.parts[-2]
|
|
|
|
pkg_dir_name = self.args.pkg_dir_name or pkg_name
|
|
update_root = "/ext/update"
|
|
flipper_update_path = f"{update_root}/{pkg_dir_name}"
|
|
|
|
self.logger.info(f'Installing "{pkg_name}" from {flipper_update_path}')
|
|
# if not os.path.exists(self.args.manifest_path):
|
|
# self.logger.error("Error: package not found")
|
|
if not self.mkdir_on_storage(
|
|
storage, update_root
|
|
) or not self.mkdir_on_storage(storage, flipper_update_path):
|
|
self.logger.error(f"Error: cannot create {storage.last_error}")
|
|
return -2
|
|
|
|
for dirpath, dirnames, filenames in os.walk(manifest_path.parents[0]):
|
|
for fname in filenames:
|
|
self.logger.debug(f"Uploading {fname}")
|
|
local_file_path = os.path.join(dirpath, fname)
|
|
flipper_file_path = f"{flipper_update_path}/{fname}"
|
|
if not self.send_file_to_storage(
|
|
storage, flipper_file_path, local_file_path, False
|
|
):
|
|
self.logger.error(f"Error: {storage.last_error}")
|
|
return -3
|
|
|
|
storage.send_and_wait_eol(
|
|
f"update install {flipper_update_path}/{manifest_name}\r"
|
|
)
|
|
result = storage.read.until(storage.CLI_EOL)
|
|
if not b"Verifying" in result:
|
|
self.logger.error(f"Unexpected response: {result.decode('ascii')}")
|
|
return -4
|
|
result = storage.read.until(storage.CLI_EOL)
|
|
if not result.startswith(b"OK"):
|
|
self.logger.error(result.decode("ascii"))
|
|
return -5
|
|
break
|
|
return 0
|
|
finally:
|
|
storage.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
Main()()
|