Add script to configure system from a JSON file

debian/bullseye-overrides/pi-top-usb-setup.lintian-overrides

@@ -1,2 +1,3 @@
 pi-top-usb-setup: no-manual-page usr/bin/pt-usb-setup
 pi-top-usb-setup: no-manual-page usr/bin/handle-usb-drive-for-pi-top-setup
+pi-top-usb-setup: no-manual-page usr/bin/pt-configure-system

debian/pi-top-usb-setup.lintian-overrides

@@ -1,2 +1,3 @@
 pi-top-usb-setup: no-manual-page [usr/bin/pt-usb-setup]
 pi-top-usb-setup: no-manual-page [usr/bin/handle-usb-drive-for-pi-top-setup]
+pi-top-usb-setup: no-manual-page [usr/bin/pt-configure-system]

debian/pt-configure-system.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=pi-top System Configuration
+Documentation=https://knowledgebase.pi-top.com/knowledge
+
+[Service]
+Type=simple
+Restart=no
+Environment="PYTHONUNBUFFERED=1"
+Environment="PYTHONDONTWRITEBYTECODE=1"
+ExecStart=/usr/bin/pt-configure-system
+TimeoutStopSec=5
+
+[Install]
+WantedBy=multi-user.target

debian/rules

@@ -9,3 +9,4 @@ override_dh_auto_test:
 
 override_dh_installsystemd:
 	dh_installsystemd --no-enable --no-start --name=pt-usb-setup@
+	dh_installsystemd --no-start --name=pt-configure-system

pi_top_usb_setup/app_fs.py

@@ -1,29 +1,11 @@
-import json
 import logging
 from pathlib import Path
 from typing import Callable, Optional
 
 from pitop.common.command_runner import run_command
-from pt_os_web_portal.backend.helpers.finalise import (
-    configure_landing,
-    deprioritise_openbox_session,
-    disable_ap_mode,
-    enable_firmware_updater_service,
-    enable_further_link_service,
-    enable_pt_miniscreen,
-    onboarding_completed,
-    restore_files,
-    stop_onboarding_autostart,
-    update_eeprom,
-)
-from pt_os_web_portal.backend.helpers.keyboard import set_keyboard_layout
-from pt_os_web_portal.backend.helpers.language import set_locale
-from pt_os_web_portal.backend.helpers.registration import set_registration_email
-from pt_os_web_portal.backend.helpers.timezone import set_timezone
-from pt_os_web_portal.backend.helpers.wifi_country import set_wifi_country
 
+from pi_top_usb_setup.configure import ConfigureSystem
 from pi_top_usb_setup.exceptions import ExtractionError, NotEnoughSpaceException
-from pi_top_usb_setup.network import Network
 from pi_top_usb_setup.utils import (
     drive_has_enough_free_space,
     extract_file,
@@ -44,7 +26,6 @@ def __init__(self, mount_point: str) -> None:
         self.USB_SETUP_FILE = f"{mount_point}/pi-top-usb-setup.tar.gz"
         self.SETUP_FOLDER = f"{self.TEMP_FOLDER}/pi-top-usb-setup"
         self.DEVICE_CONFIG = f"{self.SETUP_FOLDER}/pi-top_config.json"
-
         self.UPDATES_FOLDER = f"{self.SETUP_FOLDER}/updates"
         if get_linux_distro() == "bookworm":
             self.UPDATES_FOLDER = f"{self.SETUP_FOLDER}/updates_bookworm"
@@ -59,13 +40,14 @@ def __init__(self, mount_point: str) -> None:
                 f"Files '{self.MOUNT_POINT}/{self.USB_SETUP_FILENAME_GLOB}' or {self.UPDATES_FOLDER} not found, exiting ..."
             )
 
-        self.requires_reboot = False
         self.device = None
         if mount_point:
             self.device = run_command(
                 f"findmnt -n -o SOURCE --target {mount_point}", timeout=5
             ).strip()
 
+        self.setup = ConfigureSystem(self.DEVICE_CONFIG)
+
     def _find_setup_files(self):
         # find all setup files in mount point that match the glob pattern, sorted by date
         return sorted(
@@ -119,91 +101,11 @@ def _do_extract_setup_file(
             raise ExtractionError(f"Error extracting '{filename}': {e}")
 
     def configure_device(self, on_progress: Optional[Callable] = None) -> None:
-        if not Path(self.DEVICE_CONFIG).exists():
-            logger.info("No device configuration file found; skipping....")
-            return
-
-        def set_network(network_data):
-            # FORMAT:
-            # {
-            #     "ssid": str,
-            #     "hidden": bool
-            #     "authentication": {
-            #         "type": str,
-            #         "password": str,
-            #         ...
-            #     },
-            # }
-            logger.info(f"Setting network with data: {network_data}")
-            try:
-                Network.from_dict(network_data).connect()
-            except Exception as e:
-                logger.error(f"Error setting network: {e}")
-            return
-
-        # setting the keyboard layout requires a layout and a variant
-        lookup = {
-            "language": set_locale,
-            "country": set_wifi_country,
-            "time_zone": set_timezone,
-            "keyboard_layout": lambda layout_and_variant_arr: set_keyboard_layout(
-                *layout_and_variant_arr
-            ),
-            "email": set_registration_email,
-            "network": set_network,
-        }
-
-        logger.info(f"Configuring device using {self.DEVICE_CONFIG}")
-        with open(self.DEVICE_CONFIG) as file:
-            content = file.read()
-            config = json.loads(content)
-
-        for i, (key, function) in enumerate(lookup.items()):
-            if key not in config:
-                logger.info(f"'{key}' not found in configuration file, skipping...")
-                continue
-
-            args = config.get(key)
-            if args is None:
-                logger.info(
-                    f"Value for '{key}' not found in configuration file, skipping..."
-                )
-                continue
-
-            try:
-                logger.info(f"{key}: Executing {function} with '{args}'")
-                function(config.get(key))
-                if callable(on_progress):
-                    on_progress(float(100.0 * i / len(lookup)))
-            except Exception as e:
-                logger.error(f"{e}")
+        self.setup.configure(on_progress)
 
     def complete_onboarding(self, on_progress: Optional[Callable] = None) -> None:
-        if onboarding_completed():
-            logger.info("Device already onboarded")
-            return
-
-        functions = (
-            enable_firmware_updater_service,
-            enable_further_link_service,
-            deprioritise_openbox_session,
-            restore_files,
-            configure_landing,
-            stop_onboarding_autostart,
-            update_eeprom,
-            enable_pt_miniscreen,
-            disable_ap_mode,
-        )
+        self.setup.onboard(on_progress)
 
-        logger.info("Completing onboarding for device ...")
-        for i, func in enumerate(functions):
-            try:
-                logger.info(f"Executing {func} ...")
-                func()
-            except Exception as e:
-                logger.error(f"{e}")
-
-            if callable(on_progress):
-                on_progress(float(100.0 * i / len(functions)))
-
-        self.requires_reboot = True
+    @property
+    def requires_reboot(self) -> bool:
+        return self.setup.requires_reboot

pi_top_usb_setup/configure.py

@@ -0,0 +1,161 @@
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Callable, Dict, Optional
+
+import click
+import click_logging
+from pt_os_web_portal.backend.helpers.finalise import (
+    configure_landing,
+    deprioritise_openbox_session,
+    disable_ap_mode,
+    enable_firmware_updater_service,
+    enable_further_link_service,
+    enable_pt_miniscreen,
+    onboarding_completed,
+    restore_files,
+    stop_onboarding_autostart,
+    update_eeprom,
+)
+from pt_os_web_portal.backend.helpers.keyboard import set_keyboard_layout
+from pt_os_web_portal.backend.helpers.language import set_locale
+from pt_os_web_portal.backend.helpers.registration import set_registration_email
+from pt_os_web_portal.backend.helpers.timezone import set_timezone
+from pt_os_web_portal.backend.helpers.wifi_country import set_wifi_country
+
+from pi_top_usb_setup.network import Network
+from pi_top_usb_setup.utils import boot_partition
+
+logger = logging.getLogger()
+click_logging.basic_config(logger)
+
+
+class ConfigureSystem:
+    def __init__(self, path: str) -> None:
+        if not Path(path).exists():
+            raise FileNotFoundError(
+                f"Configuration file '{path}' doesn't exist; exiting ..."
+            )
+        self.path = path
+        self.config: Dict = {}
+        self.read()
+        self.requires_reboot = False
+
+    def read(self):
+        with open(self.path) as file:
+            content = file.read()
+            self.config = json.loads(content)
+
+    def configure(self, on_progress: Optional[Callable] = None) -> None:
+        def set_network(network_data):
+            logger.info(f"Setting network with data: {network_data}")
+            try:
+                Network.from_dict(network_data).connect()
+            except Exception as e:
+                logger.error(f"Error setting network: {e}")
+            return
+
+        # setting the keyboard layout requires a layout and a variant
+        lookup = {
+            "language": set_locale,
+            "country": set_wifi_country,
+            "time_zone": set_timezone,
+            "keyboard_layout": lambda layout_and_variant_arr: set_keyboard_layout(
+                *layout_and_variant_arr
+            ),
+            "email": set_registration_email,
+            "network": set_network,
+        }
+
+        for i, (key, function) in enumerate(lookup.items()):
+            if key not in self.config:
+                logger.info(f"'{key}' not found in configuration file, skipping...")
+                continue
+
+            args = self.config.get(key)
+            if args is None:
+                logger.info(
+                    f"Value for '{key}' not found in configuration file, skipping..."
+                )
+                continue
+
+            try:
+                logger.info(f"{key}: Executing {function} with '{args}'")
+                function(self.config.get(key))
+                if callable(on_progress):
+                    on_progress(float(100.0 * i / len(lookup)))
+            except Exception as e:
+                logger.error(f"{e}")
+
+    def onboard(
+        self,
+        on_progress: Optional[Callable] = None,
+    ) -> None:
+        if onboarding_completed():
+            logger.info("Device already onboarded, skipping...")
+            return
+
+        functions = (
+            enable_firmware_updater_service,
+            enable_further_link_service,
+            deprioritise_openbox_session,
+            restore_files,
+            configure_landing,
+            stop_onboarding_autostart,
+            update_eeprom,
+            enable_pt_miniscreen,
+            disable_ap_mode,
+        )
+
+        logger.info("Completing onboarding for device ...")
+        for i, func in enumerate(functions):
+            try:
+                logger.info(f"Executing {func} ...")
+                func()
+            except Exception as e:
+                logger.error(f"{e}")
+
+            if callable(on_progress):
+                on_progress(float(100.0 * i / len(functions)))
+
+        self.requires_reboot = True
+
+    def run_all(self, on_progress: Optional[Callable] = None):
+        self.configure(on_progress)
+        self.onboard(on_progress)
+
+
+def is_root():
+    return os.geteuid() == 0
+
+
+@click.command()
+@click_logging.simple_verbosity_option(logger)
+@click.version_option()
+@click.argument("path_to_json", required=False)
+def main(path_to_json) -> None:
+    if not is_root():
+        logger.error("This script must be run as root")
+        return
+
+    if path_to_json is None:
+        path_to_json = f"{boot_partition()}/pi-top_config.json"
+        logger.info(f"Using path to json: {path_to_json}")
+
+    try:
+        c = ConfigureSystem(path_to_json)
+        c.run_all()
+
+        logger.info(
+            f"Configuration completed - removing configuration file {path_to_json}"
+        )
+        Path(path_to_json).unlink(missing_ok=True)
+
+        if c.requires_reboot:
+            logger.info("Rebooting ...")
+            os.system("reboot")
+    except FileNotFoundError:
+        logger.info(f"Configuration file {path_to_json} not found, skipping ...")
+    except Exception as e:
+        logger.error(f"{e}")

pi_top_usb_setup/utils.py

@@ -189,8 +189,14 @@ def restart_service_and_skip_user_confirmation_dialog(mount_point: str):
     os.kill(os.getpid(), signal.SIGINT)
 
 
-def get_linux_distro():
+def get_linux_distro() -> str:
     cmd = "grep VERSION_CODENAME /etc/os-release | cut -d'=' -f2"
     process = Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE, stderr=PIPE)
     stdout, _ = process.communicate()
     return stdout.decode().strip()
+
+
+def boot_partition() -> str:
+    if get_linux_distro() == "bookworm":
+        return "/boot/firmware"
+    return "/boot"

setup.cfg

@@ -30,6 +30,7 @@ install_requires =
 [options.entry_points]
 console_scripts =
     pt-usb-setup = pi_top_usb_setup.__main__:main
+    pt-configure-system = pi_top_usb_setup.configure:main
 
 [bdist_wheel]
 universal = 1