Add Private Credentials to a Private Repository (#166)

* Unified type of `args.magnify_font_sizes`.

* Added `run()`. (#162)

* Migrated main body to `run()`. (#162)

* Normalized imports. (#162)

Author: ATATC

leads_vec/__init__.py

@@ -2,3 +2,5 @@
 
 if not _find_spec("pynput"):
     raise ImportError("Please install `pynput` to run this module\n>>>pip install pynput")
+
+from leads_vec.run import *

leads_vec/__main__.py

@@ -1,36 +1,30 @@
 from argparse import ArgumentParser as _ArgumentParser, BooleanOptionalAction as _BooleanOptionalAction
 from importlib.metadata import version as _package_version, PackageNotFoundError as _PackageNotFoundError
-from importlib.util import spec_from_file_location as _spec_from_file_location, module_from_spec as _module_from_spec
 from os import getlogin as _get_login
 from os.path import abspath as _abspath
-from os.path import exists as _exists
 from sys import exit as _exit, version as _version
 from warnings import filterwarnings as _filterwarnings
 
-from customtkinter import set_default_color_theme as _set_default_color_theme
-
-from leads import register_controller as _register_controller, MAIN_CONTROLLER as _MAIN_CONTROLLER, \
-    L as _L, load_config as _load_config, register_config as _register_config, reset as _reset
-from leads.data_persistence import Dataset as _Dataset
-from leads_gui import Config as _Config
+from leads import L as _L
 from leads_gui.system import get_system_kernel as _get_system_kernel
+from leads_vec.run import run
 
 if __name__ == "__main__":
     _filterwarnings("ignore")
-    _MODULE_PATH = _abspath(__file__)[:-12]
+    MODULE_PATH = _abspath(__file__)[:-12]
 
     parser = _ArgumentParser(prog="LEADS VeC",
                              description="Lightweight Embedded Assisted Driving System VeC",
                              epilog="ProjectNeura: https://projectneura.org\n"
                                     "GitHub: https://github.com/ProjectNeura/LEADS")
     parser.add_argument("action", choices=("info", "replay", "run"))
     parser.add_argument("-c", "--config", default=None, help="specify a configuration file")
-    parser.add_argument("-d", "--devices", default=f"{_MODULE_PATH}/devices.py",
+    parser.add_argument("-d", "--devices", default=f"{MODULE_PATH}/devices.py",
                         help="specify a devices module")
     parser.add_argument("-r", "--register", choices=("systemd", "config", "reverse_proxy"), default=None,
                         help="register a service")
     parser.add_argument("-t", "--theme", default=None, help="specify a theme")
-    parser.add_argument("-mfs", "--magnify-font-sizes", type=float, default=None, help="magnify font sizes by a factor")
+    parser.add_argument("-mfs", "--magnify-font-sizes", type=float, default=1, help="magnify font sizes by a factor")
     parser.add_argument("--emu", action=_BooleanOptionalAction, default=False, help="use emulator")
     parser.add_argument("--auto-mfs", action=_BooleanOptionalAction, default=False,
                         help="automatically magnify font sizes to match the original proportion")
@@ -51,68 +45,15 @@
                 f"Python Version: {_version}",
                 f"User: {_get_login()}",
                 f"`frpc` Available: {_frpc_exists()}",
-                f"Module Path: {_MODULE_PATH}",
+                f"Module Path: {MODULE_PATH}",
                 f"LEADS Version: {leads_version}",
                 f"LEADS VeC Version: {__version__}",
                 sep="\n")
-        _exit()
-    if args.register == "systemd":
-        from ._bootloader import register_leads_vec as _create_service
-
-        _create_service()
-        _L.info("Service registered")
-        _L.info(f"Service script is located at \"{_MODULE_PATH}/_bootloader/leads-vec.service.sh\"")
-    elif args.register == "config":
-        if _exists("config.json"):
-            r = input("\"config.json\" already exists. Overwrite? (y/N) >>>").lower()
-            if r.lower() != "y":
-                _exit("Error: Aborted")
-        with open("config.json", "w") as f:
-            f.write(str(_Config({})))
-        _L.info("Configuration file saved to \"config.json\"")
-    elif args.register == "reverse_proxy":
-        from ._bootloader import start_frpc as _start_frpc
-
-        _start_frpc()
-        _L.info("`frpc` started")
-    config = _load_config(args.config, _Config) if args.config else _Config({})
-    _L.debug("Configuration loaded:", str(config))
-    if t := args.theme:
-        _set_default_color_theme(t)
-    if f := args.magnify_font_sizes:
-        config.magnify_font_sizes(f)
-    if args.auto_mfs:
-        config.auto_magnify_font_sizes()
-    _register_config(config)
-    from leads_vec.cli import main
-
-    if args.action == "replay":
-        from leads_emulation.replay import ReplayController as _Controller
-
-        _register_controller(_MAIN_CONTROLLER, _Controller(_Dataset(f"{config.data_dir}/main.csv")))
-        _L.info("Replay started")
-        _exit(main())
-
-    try:
-        if args.emu:
-            raise SystemError("User specifies to use emulator")
-        spec = _spec_from_file_location("_", args.devices)
-        spec.loader.exec_module(_module_from_spec(spec))
-    except (ImportError, SystemError) as e:
-        _L.debug(repr(e))
-        if isinstance(e, ImportError):
-            if args.ignore_import_error:
-                _L.debug(f"Ignoring import error: {repr(e)}")
-                _exit(main())
-            else:
-                _L.warn(
-                    f"Specified devices module ({args.devices}) is not available, using emulation module instead...")
-        _reset()
-        try:
-            from leads_emulation import SinController as _Controller
-
-            _register_controller(_MAIN_CONTROLLER, _Controller(0, 200, .05))
-        except ImportError as e:
-            _L.error(f"Emulator error: {repr(e)}")
-            _exit(1)
-    _exit(main())
+    else:
+        if args.action == "replay":
+            args.devices = f"{MODULE_PATH}/replay.py"
+            args.emu = False
+            _L.debug("Replay mode enabled")
+        _exit(run(args.config, args.devices, args.register, args.theme, args.magnify_font_sizes, args.emu,
+                  args.auto_mfs, args.ignore_import_error))
+    _exit()

leads_vec/cli.py

@@ -6,11 +6,16 @@
     CTkSegmentedButton as _CTkSegmentedButton
 from pynput.keyboard import Listener as _Listener, Key as _Key, KeyCode as _KeyCode
 
-from leads import *
-from leads.comm import *
-from leads_audio import *
-from leads_gui import *
-from leads_raspberry_pi import *
+from leads import LEADS, SystemLiteral, require_config, register_context, DTCS, ABS, EBI, ATBS, GPSSpeedCorrection, \
+    ESCMode, get_controller, MAIN_CONTROLLER, L, EventListener, DataPushedEvent, UpdateEvent, has_device, \
+    GPS_RECEIVER, get_device, InterventionEvent, SuspensionEvent, Event, LEFT_INDICATOR, RIGHT_INDICATOR, SFT, \
+    initialize_main
+from leads.comm import Callback, Service, start_server, create_server, my_ip_addresses
+from leads_audio import DIRECTION_INDICATOR_ON, DIRECTION_INDICATOR_OFF, WARNING, CONFIRM
+from leads_gui import RuntimeData, Window, GForceVar, FrequencyGenerator, Left, Color, Right, ContextManager, \
+    Typography, Speedometer, ProxyCanvas, SpeedTrendMeter, GForceMeter, Stopwatch, Hazard, initialize, Battery, Brake, \
+    ESC, Satellite, Motor, Speed
+from leads_raspberry_pi import LEDGroupCommand, LEDCommand, Transition, Entire
 from leads_vec.__version__ import __version__
 
 

leads_vec/replay.py

@@ -0,0 +1,8 @@
+from leads import controller, MAIN_CONTROLLER, require_config
+from leads.data_persistence import Dataset
+from leads_emulation.replay import ReplayController
+
+
+@controller(MAIN_CONTROLLER, args=(Dataset(f"{require_config().data_dir}/main.csv"),))
+class MainController(ReplayController):
+    pass

leads_vec/run.py

@@ -0,0 +1,73 @@
+from importlib.util import spec_from_file_location as _spec_from_file_location, module_from_spec as _module_from_spec
+from os.path import abspath as _abspath
+from os.path import exists as _exists
+from typing import Literal as _Literal
+
+from customtkinter import set_default_color_theme as _set_default_color_theme
+
+from leads import register_controller as _register_controller, MAIN_CONTROLLER as _MAIN_CONTROLLER, \
+    L as _L, load_config as _load_config, register_config as _register_config, reset as _reset
+from leads_gui import Config as _Config
+
+
+def run(config: str,
+        devices: str,
+        register: _Literal["systemd", "config", "reverse_proxy"],
+        theme: str,
+        magnify_font_sizes: float,
+        emu: bool,
+        auto_mfs: bool,
+        ignore_import_error: bool) -> int:
+    if register == "systemd":
+        from ._bootloader import register_leads_vec as _create_service
+
+        _create_service()
+        _L.debug("Service registered")
+        _L.debug(f"Service script is located at \"{_abspath(__file__)[:-6]}_bootloader/leads-vec.service.sh\"")
+    elif register == "config":
+        if _exists("config.json"):
+            r = input("\"config.json\" already exists. Overwrite? (Y/n) >>>").lower()
+            if r.lower() != "y":
+                _L.error("Aborted")
+                return 1
+        with open("config.json", "w") as f:
+            f.write(str(_Config({})))
+        _L.debug("Configuration file saved to \"config.json\"")
+    elif register == "reverse_proxy":
+        from ._bootloader import start_frpc as _start_frpc
+
+        _start_frpc()
+        _L.debug("`frpc` started")
+    config = _load_config(config, _Config) if config else _Config({})
+    _L.debug("Configuration loaded:", str(config))
+    if t := theme:
+        _set_default_color_theme(t)
+    if (f := magnify_font_sizes) != 1:
+        config.magnify_font_sizes(f)
+    if auto_mfs:
+        config.auto_magnify_font_sizes()
+    _register_config(config)
+    from leads_vec.cli import main
+
+    try:
+        if emu:
+            raise SystemError("User specifies to use emulator")
+        spec = _spec_from_file_location("_", devices)
+        spec.loader.exec_module(_module_from_spec(spec))
+    except (ImportError, SystemError) as e:
+        _L.debug(repr(e))
+        if isinstance(e, ImportError):
+            if ignore_import_error:
+                _L.debug(f"Ignoring import error: {repr(e)}")
+                return main()
+            else:
+                _L.warn(f"Specified devices module ({devices}) is not available, using emulation module instead...")
+        _reset()
+        try:
+            from leads_emulation import SinController as _Controller
+
+            _register_controller(_MAIN_CONTROLLER, _Controller(0, 200, .05))
+        except ImportError as e:
+            _L.error(f"Emulator error: {repr(e)}")
+            return 1
+    return main()