Flet CLI moved into separate module

sdk/python/flet/cli/cli.py

@@ -0,0 +1,79 @@
+import argparse
+import sys
+import flet.version
+import flet.cli.commands.run
+import flet.cli.commands.build
+
+# Source https://stackoverflow.com/a/26379693
+def set_default_subparser(self, name, args=None, positional_args=0):
+    """default subparser selection. Call after setup, just before parse_args()
+    name: is the name of the subparser to call by default
+    args: if set is the argument list handed to parse_args()
+
+    , tested with 2.7, 3.2, 3.3, 3.4
+    it works with 2.6 assuming argparse is installed
+    """
+    subparser_found = False
+    existing_default = False  # check if default parser previously defined
+    for arg in sys.argv[1:]:
+        if arg in ["-h", "--help", "--version"]:  # global help if no subparser
+            break
+    else:
+        for x in self._subparsers._actions:
+            if not isinstance(x, argparse._SubParsersAction):
+                continue
+            for sp_name in x._name_parser_map.keys():
+                if sp_name in sys.argv[1:]:
+                    subparser_found = True
+                if sp_name == name:  # check existance of default parser
+                    existing_default = True
+        if not subparser_found:
+            # If the default subparser is not among the existing ones,
+            # create a new parser.
+            # As this is called just before 'parse_args', the default
+            # parser created here will not pollute the help output.
+
+            if not existing_default:
+                for x in self._subparsers._actions:
+                    if not isinstance(x, argparse._SubParsersAction):
+                        continue
+                    x.add_parser(name)
+                    break  # this works OK, but should I check further?
+
+            # insert default in last position before global positional
+            # arguments, this implies no global options are specified after
+            # first positional argument
+            if args is None and len(sys.argv) > 1:
+                sys.argv.insert(positional_args, name)
+            elif args is not None:
+                args.insert(positional_args, name)
+            # print(sys.argv)
+
+
+argparse.ArgumentParser.set_default_subparser = set_default_subparser
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--version", action="version", version=flet.version.version)
+    sp = parser.add_subparsers(dest="command")
+    # sp.default = "run"
+
+    flet.cli.commands.run.Command.register_to(sp, "run")
+    flet.cli.commands.build.Command.register_to(sp, "build")
+    parser.set_default_subparser("run", positional_args=1)
+
+    # print usage if called without args
+    if len(sys.argv) == 1:
+        parser.print_help(sys.stdout)
+        sys.exit(1)
+
+    # parse args
+    args = parser.parse_args()
+
+    # execute command
+    args.handler(args)
+
+
+if __name__ == "__main__":
+    main()

sdk/python/flet/cli/commands/base.py

@@ -0,0 +1,57 @@
+import argparse
+from typing import Any, List, Optional
+
+from flet.cli.commands.options import Option, verbose_option
+
+
+class BaseCommand:
+    """A CLI subcommand"""
+
+    # The subcommand's name
+    name: Optional[str] = None
+    # The subcommand's help string, if not given, __doc__ will be used.
+    description: Optional[str] = None
+    # A list of pre-defined options which will be loaded on initializing
+    # Rewrite this if you don't want the default ones
+    arguments: List[Option] = [verbose_option]
+
+    def __init__(self, parser: argparse.ArgumentParser) -> None:
+        for arg in self.arguments:
+            arg.add_to_parser(parser)
+        self.add_arguments(parser)
+
+    @classmethod
+    def register_to(
+        cls,
+        subparsers: argparse._SubParsersAction,
+        name: Optional[str] = None,
+        **kwargs: Any
+    ) -> None:
+        """Register a subcommand to the subparsers,
+        with an optional name of the subcommand.
+        """
+        help_text = cls.description or cls.__doc__
+        name = name or cls.name or ""
+        # Remove the existing subparser as it will raises an error on Python 3.11+
+        subparsers._name_parser_map.pop(name, None)
+        subactions = subparsers._get_subactions()
+        subactions[:] = [action for action in subactions if action.dest != name]
+        parser = subparsers.add_parser(
+            name,
+            description=help_text,
+            help=help_text,
+            # formatter_class=PdmFormatter,
+            **kwargs,
+        )
+        command = cls(parser)
+        parser.set_defaults(handler=command.handle)
+
+    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
+        """Manipulate the argument parser to add more arguments"""
+        pass
+
+    def handle(self, options: argparse.Namespace) -> None:
+        """The command handler function.
+        :param options: the parsed Namespace object
+        """
+        raise NotImplementedError

sdk/python/flet/cli/commands/build.py

@@ -0,0 +1,23 @@
+import argparse
+from flet.cli.commands.base import BaseCommand
+
+
+class Command(BaseCommand):
+    """
+    Package Flet app to a standalone bundle
+    """
+
+    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument("script", type=str, help="path to a Python script")
+        parser.add_argument(
+            "-F",
+            "--onefile",
+            dest="onefile",
+            action="store_true",
+            default=False,
+            help="create a one-file bundled executable",
+        )
+
+    def handle(self, options: argparse.Namespace) -> None:
+        # print("BUILD COMMAND", options)
+        raise NotImplementedError("Build command is not implemented yet.")

sdk/python/flet/cli/commands/options.py

@@ -0,0 +1,27 @@
+import argparse
+from typing import Any
+
+
+class Option:
+    """A reusable option object which delegates all arguments
+    to parser.add_argument().
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.args = args
+        self.kwargs = kwargs
+
+    def add_to_parser(self, parser: argparse._ActionsContainer) -> None:
+        parser.add_argument(*self.args, **self.kwargs)
+
+    def add_to_group(self, group: argparse._ArgumentGroup) -> None:
+        group.add_argument(*self.args, **self.kwargs)
+
+
+verbose_option = Option(
+    "-v",
+    "--verbose",
+    action="count",
+    default=0,
+    help="-v for detailed output and -vv for more detailed",
+)

sdk/python/flet/cli/commands/run.py

@@ -0,0 +1,178 @@
+import argparse
+import logging
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import threading
+import time
+from flet.cli.commands.base import BaseCommand
+from flet.flet import open_flet_view
+from flet.utils import get_free_tcp_port, is_windows, open_in_browser
+from watchdog.events import FileSystemEventHandler
+from watchdog.observers import Observer
+
+
+class Command(BaseCommand):
+    """
+    Run Flet app
+    """
+
+    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
+        parser.add_argument("script", type=str, help="path to a Python script")
+        parser.add_argument(
+            "-p",
+            "--port",
+            dest="port",
+            type=int,
+            default=None,
+            help="custom TCP port to run Flet app on",
+        )
+        parser.add_argument(
+            "-d",
+            "--directory",
+            dest="directory",
+            action="store_true",
+            default=False,
+            help="watch script directory",
+        )
+        parser.add_argument(
+            "-r",
+            "--recursive",
+            dest="recursive",
+            action="store_true",
+            default=False,
+            help="watch script directory and all sub-directories recursively",
+        )
+        parser.add_argument(
+            "-n",
+            "--hidden",
+            dest="hidden",
+            action="store_true",
+            default=False,
+            help="application window is hidden on startup",
+        )
+        parser.add_argument(
+            "-w",
+            "--web",
+            dest="web",
+            action="store_true",
+            default=False,
+            help="open app in a web browser",
+        )
+
+    def handle(self, options: argparse.Namespace) -> None:
+        # print("RUN COMMAND", options)
+        script_path = options.script
+        if not os.path.isabs(options.script):
+            script_path = str(Path(os.getcwd()).joinpath(options.script).resolve())
+
+        if not Path(script_path).exists():
+            print(f"File not found: {script_path}")
+            exit(1)
+
+        script_dir = os.path.dirname(script_path)
+
+        port = options.port
+        if options.port is None:
+            port = get_free_tcp_port()
+
+        my_event_handler = Handler(
+            [sys.executable, "-u", script_path],
+            None if options.directory or options.recursive else script_path,
+            port,
+            options.web,
+            options.hidden,
+        )
+
+        my_observer = Observer()
+        my_observer.schedule(my_event_handler, script_dir, recursive=options.recursive)
+        my_observer.start()
+
+        try:
+            while True:
+                if my_event_handler.terminate.wait(1):
+                    break
+        except KeyboardInterrupt:
+            pass
+
+        if my_event_handler.fvp is not None and not is_windows():
+            try:
+                logging.debug(f"Flet View process {my_event_handler.fvp.pid}")
+                os.kill(my_event_handler.fvp.pid + 1, signal.SIGKILL)
+            except:
+                pass
+        my_observer.stop()
+        my_observer.join()
+
+
+class Handler(FileSystemEventHandler):
+    def __init__(self, args, script_path, port, web, hidden) -> None:
+        super().__init__()
+        self.args = args
+        self.script_path = script_path
+        self.port = port
+        self.web = web
+        self.hidden = hidden
+        self.last_time = time.time()
+        self.is_running = False
+        self.fvp = None
+        self.page_url_prefix = f"PAGE_URL_{time.time()}"
+        self.page_url = None
+        self.terminate = threading.Event()
+        self.start_process()
+
+    def start_process(self):
+        p_env = {**os.environ}
+        if self.port is not None:
+            p_env["FLET_SERVER_PORT"] = str(self.port)
+        p_env["FLET_DISPLAY_URL_PREFIX"] = self.page_url_prefix
+
+        self.p = subprocess.Popen(self.args, env=p_env, stdout=subprocess.PIPE)
+        self.is_running = True
+        th = threading.Thread(target=self.print_output, args=[self.p], daemon=True)
+        th.start()
+
+    def on_any_event(self, event):
+        if (
+            self.script_path is None or event.src_path == self.script_path
+        ) and not event.is_directory:
+            current_time = time.time()
+            if (current_time - self.last_time) > 0.5 and self.is_running:
+                self.last_time = current_time
+                th = threading.Thread(target=self.restart_program, args=(), daemon=True)
+                th.start()
+
+    def print_output(self, p):
+        while True:
+            line = p.stdout.readline()
+            if not line:
+                break
+            line = line.decode("utf-8").rstrip("\r\n")
+            if line.startswith(self.page_url_prefix):
+                if not self.page_url:
+                    self.page_url = line[len(self.page_url_prefix) + 1 :]
+                    print(self.page_url)
+                    if self.web:
+                        open_in_browser(self.page_url)
+                    else:
+                        th = threading.Thread(
+                            target=self.open_flet_view_and_wait, args=(), daemon=True
+                        )
+                        th.start()
+            else:
+                print(line)
+
+    def open_flet_view_and_wait(self):
+        self.fvp = open_flet_view(self.page_url, self.hidden)
+        self.fvp.wait()
+        self.p.kill()
+        self.terminate.set()
+
+    def restart_program(self):
+        self.is_running = False
+        self.p.kill()
+        self.p.wait()
+        time.sleep(0.5)
+        self.start_process()