fbt: fixes (#1352)

* fbt: added --git-tasks; fixed typos
* fbt: fixed --extra-int-apps handling; scripts: moved storage.py & selfupdate.py to App() framework
* fbt: changed pseudo-builders to PhonyTargets with commands; added link to latest build dir as build/latest
* fbt: Restored old ep git handling
* fbt: dropped git tasks & dirlink.py
* fbt: removed extra quoting in fbt.cmd
* docs: added flash_usb to ReadMe.md

Co-authored-by: あく <alleteam@gmail.com>

scripts/flipper/app.py

@@ -1,7 +1,6 @@
 import logging
 import argparse
 import sys
-import os
 
 
 class App:

scripts/flipper/utils/cdc.py

@@ -0,0 +1,17 @@
+import serial.tools.list_ports as list_ports
+
+# Returns a valid port or None, if it cannot be found
+def resolve_port(logger, portname: str = "auto"):
+    if portname != "auto":
+        return portname
+    # Try guessing
+    flippers = list(list_ports.grep("flip"))
+    if len(flippers) == 1:
+        flipper = flippers[0]
+        logger.info(f"Using {flipper.serial_number} on {flipper.device}")
+        return flipper.device
+    elif len(flippers) == 0:
+        logger.error("Failed to find connected Flipper")
+    elif len(flippers) > 1:
+        logger.error("More than one Flipper is attached")
+    logger.error("Failed to guess which port to use. Specify --port")

scripts/selfupdate.py

@@ -1,20 +1,18 @@
 #!/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 argparse
 import os
-import sys
 import pathlib
 import serial.tools.list_ports as list_ports
 
 
-class Main:
-    def __init__(self):
-        # command args
-        self.parser = argparse.ArgumentParser()
-        self.parser.add_argument("-d", "--debug", action="store_true", help="Debug")
+class Main(App):
+    def init(self):
         self.parser.add_argument("-p", "--port", help="CDC Port", default="auto")
         self.parser.add_argument(
             "-b",
@@ -25,35 +23,15 @@ class Main:
             type=int,
         )
 
-        self.subparsers = self.parser.add_subparsers(help="sub-command help")
-
-        self.parser_install = self.subparsers.add_parser(
-            "install", help="Install OTA package"
-        )
-        self.parser_install.add_argument("manifest_path", help="Manifest path")
-        self.parser_install.add_argument(
+        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_install.set_defaults(func=self.install)
+        self.parser.set_defaults(func=self.install)
 
         # 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()
-
     # make directory with exist check
     def mkdir_on_storage(self, storage, flipper_dir_path):
         if not storage.exist_dir(flipper_dir_path):
@@ -82,61 +60,49 @@ class Main:
                 return False
         return True
 
-    def _get_port(self):
-        if self.args.port != "auto":
-            return self.args.port
-        # Try guessing
-        flippers = list(list_ports.grep("flip"))
-        if len(flippers) == 1:
-            flipper = flippers[0]
-            self.logger.info(f"Using {flipper.serial_number} on {flipper.device}")
-            return flipper.device
-        elif len(flippers) == 0:
-            self.logger.error("Failed to find connected Flipper")
-        elif len(flippers) > 1:
-            self.logger.error("More than one Flipper is attached")
-        self.logger.error("Failed to guess which port to use. Specify --port")
-
     def install(self):
-        if not (port := self._get_port()):
+        if not (port := resolve_port(self.logger, self.args.port)):
             return 1
 
         storage = FlipperStorage(port, self.args.baud)
         storage.start()
 
-        if not os.path.isfile(self.args.manifest_path):
-            self.logger.error("Error: manifest not found")
-            return 2
+        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]
+            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
-        flipper_update_path = f"/ext/update/{pkg_dir_name}"
+            pkg_dir_name = self.args.pkg_dir_name or pkg_name
+            flipper_update_path = f"/ext/update/{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, flipper_update_path):
-            self.logger.error(f"Error: cannot create {storage.last_error}")
-            return -2
+            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, 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
+            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"
-            )
-            break
-        storage.stop()
+                storage.send_and_wait_eol(
+                    f"update install {flipper_update_path}/{manifest_name}\r"
+                )
+                break
+            return 0
+        finally:
+            storage.stop()
 
 
 if __name__ == "__main__":

scripts/storage.py

@@ -1,23 +1,19 @@
 #!/usr/bin/env python3
 
+from flipper.app import App
 from flipper.storage import FlipperStorage
+from flipper.utils.cdc import resolve_port
 
 import logging
-import argparse
 import os
-import sys
 import binascii
-import posixpath
 import filecmp
 import tempfile
 
 
-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)
+class Main(App):
+    def init(self):
+        self.parser.add_argument("-p", "--port", help="CDC Port", default="auto")
         self.parser.add_argument(
             "-b",
             "--baud",
@@ -77,43 +73,37 @@ class Main:
         )
         self.parser_stress.set_defaults(func=self.stress)
 
-        # logging
-        self.logger = logging.getLogger()
+    def _get_storage(self):
+        if not (port := resolve_port(self.logger, self.args.port)):
+            return None
 
-    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()
+        storage = FlipperStorage(port, self.args.baud)
+        storage.start()
+        return storage
 
     def mkdir(self):
-        storage = FlipperStorage(self.args.port)
-        storage.start()
+        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
 
     def remove(self):
-        storage = FlipperStorage(self.args.port)
-        storage.start()
+        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
 
     def receive(self):
-        storage = FlipperStorage(self.args.port)
-        storage.start()
+        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):
@@ -155,14 +145,17 @@ class Main:
             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
 
     def send(self):
-        storage = FlipperStorage(self.args.port)
-        storage.start()
+        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):
@@ -250,8 +243,9 @@ class Main:
                     self.logger.error(f"Error: {storage.last_error}")
 
     def read(self):
-        storage = FlipperStorage(self.args.port, self.args.baud)
-        storage.start()
+        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:
@@ -264,10 +258,12 @@ class Main:
                 print("Binary hexadecimal data:")
                 print(binascii.hexlify(data).decode())
         storage.stop()
+        return 0
 
     def size(self):
-        storage = FlipperStorage(self.args.port)
-        storage.start()
+        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:
@@ -275,13 +271,16 @@ class Main:
         else:
             print(size)
         storage.stop()
+        return 0
 
     def list(self):
-        storage = FlipperStorage(self.args.port)
-        storage.start()
+        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
 
     def stress(self):
         self.logger.error("This test is wearing out flash memory.")
@@ -293,18 +292,21 @@ class Main:
             self.logger.error("Stop at this point or device warranty will be void")
             say = input("Anything to say? ").strip().lower()
             if say != "void":
-                return
+                return 2
             say = input("Why, Mr. Anderson? ").strip().lower()
             if say != "because":
-                return
+                return 3
 
         with tempfile.TemporaryDirectory() as tmpdirname:
             send_file_name = os.path.join(tmpdirname, "send")
             receive_file_name = os.path.join(tmpdirname, "receive")
             with open(send_file_name, "w") as fout:
                 fout.write("A" * self.args.file_size)
-            storage = FlipperStorage(self.args.port)
-            storage.start()
+
+            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
@@ -318,6 +320,7 @@ class Main:
                 os.unlink(receive_file_name)
                 self.args.count -= 1
             storage.stop()
+            return 0
 
 
 if __name__ == "__main__":