mirror of
https://github.com/DarkFlippers/unleashed-firmware.git
synced 2024-12-27 23:33:53 +03:00
106 lines
3.5 KiB
Python
Executable File
106 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import operator
|
|
from functools import reduce
|
|
import time
|
|
|
|
from flipper.app import App
|
|
from flipper.storage import FlipperStorage, FlipperStorageOperations
|
|
from flipper.utils.cdc import resolve_port
|
|
|
|
|
|
class Main(App):
|
|
APP_POST_CLOSE_DELAY_SEC = 0.2
|
|
|
|
def init(self):
|
|
self.parser.add_argument("-p", "--port", help="CDC Port", default="auto")
|
|
self.parser.add_argument(
|
|
"--sources",
|
|
"-s",
|
|
nargs="+",
|
|
action="append",
|
|
default=[],
|
|
help="Files to send",
|
|
)
|
|
self.parser.add_argument(
|
|
"--targets",
|
|
"-t",
|
|
nargs="+",
|
|
action="append",
|
|
default=[],
|
|
help="File destinations (must be same length as -s)",
|
|
)
|
|
self.parser.add_argument(
|
|
"--host-app",
|
|
"-a",
|
|
help="Host app to launch",
|
|
)
|
|
|
|
self.parser.set_defaults(func=self.install)
|
|
|
|
@staticmethod
|
|
def flatten(item_list):
|
|
return reduce(operator.concat, item_list, [])
|
|
|
|
def install(self):
|
|
self.args.sources = self.flatten(self.args.sources)
|
|
self.args.targets = self.flatten(self.args.targets)
|
|
|
|
if len(self.args.sources) != len(self.args.targets):
|
|
self.logger.error(
|
|
f"Error: sources ({self.args.sources}) and targets ({self.args.targets}) must be same length"
|
|
)
|
|
return 1
|
|
|
|
if not (port := resolve_port(self.logger, self.args.port)):
|
|
return 2
|
|
|
|
try:
|
|
with FlipperStorage(port) as storage:
|
|
storage_ops = FlipperStorageOperations(storage)
|
|
for fap_local_path, fap_dst_path in zip(
|
|
self.args.sources, self.args.targets
|
|
):
|
|
self.logger.info(f'Installing "{fap_local_path}" to {fap_dst_path}')
|
|
|
|
storage_ops.recursive_send(fap_dst_path, fap_local_path, False)
|
|
|
|
fap_host_app = self.args.targets[0]
|
|
startup_command = f"{fap_host_app}"
|
|
if self.args.host_app:
|
|
startup_command = self.args.host_app
|
|
|
|
self.logger.info("Closing current app, if any")
|
|
for _ in range(10):
|
|
storage.send_and_wait_eol("loader close\r")
|
|
result = storage.read.until(storage.CLI_EOL)
|
|
if b"was closed" in result:
|
|
self.logger.info("App closed")
|
|
storage.read.until(storage.CLI_EOL)
|
|
time.sleep(self.APP_POST_CLOSE_DELAY_SEC)
|
|
elif result.startswith(b"No application"):
|
|
storage.read.until(storage.CLI_EOL)
|
|
break
|
|
else:
|
|
self.logger.error(
|
|
f"Unexpected response: {result.decode('ascii')}"
|
|
)
|
|
return 4
|
|
|
|
self.logger.info(f"Launching app: {startup_command}")
|
|
storage.send_and_wait_eol(f'loader open "{startup_command}"\r')
|
|
|
|
if len(result := storage.read.until(storage.CLI_EOL)):
|
|
self.logger.error(f"Unexpected response: {result.decode('ascii')}")
|
|
return 3
|
|
return 0
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error: {e}")
|
|
# raise
|
|
return 4
|
|
|
|
|
|
if __name__ == "__main__":
|
|
Main()()
|