Builder refactor with Python and the ability to layer builds

Makefile

@@ -1,7 +1,9 @@
 .PHONY: focal-rt-ros2 clean
 
+# TODO: eventually the build.py should be a command line script that takes
+#       arguments
 focal-rt-ros2:
-	sudo builder/main.sh focal-rt-ros2/vars.sh
+	sudo python3 build.py
 
 clean:
 	sudo rm -rf out cache

focal-rt-ros2/config.ini

@@ -22,7 +22,7 @@ qemu_user_static_path = /usr/bin/qemu-aarch64-static
 
 # Uncomment this if you want to pause the builder after a particular stage to
 # debug/experiment.
-# pause_after = download_and_extract_image_if_necessary
+# pause_after = cleanup_chroot
 
 # This section contains environment variables that will be exported to the
 # phase1/phase2 build scripts.

image_builder/builder.py

@@ -124,18 +124,16 @@ def verify_build_can_proceed(self):
       if shutil.which(command) is None:
         raise RequirementNotMetError(f"command {command} is not found on the host system but is required")
 
-  def _parse_config(self, filename: str) -> tuple[dict, dict]:
-    config = configparser.ConfigParser()
-    # Preserve case sensitivity for configuration keys so that environment variables are properly exported.
-    config.optionxform = str
-    config.read(filename)
-    return (
-      dict(config["build"].items()),
-      dict(config["env"].items()),
-    )
+  @property
+  def loop_device(self):
+    if getattr(self, "_loop_device", None) is None:
+      with open(self.session_loop_device_file) as f:
+        self._loop_device = f.read().strip()
+
+    return self._loop_device
 
   def build(self):
-    self.log_builder_information()
+    self._log_builder_information()
 
     os.makedirs(self.cache_dir, exist_ok=True)
     os.chmod(self.cache_dir, 0o777)
@@ -163,38 +161,6 @@ def build(self):
     self.logger.info(msg)
     self.logger.info("-" * len(msg))
 
-  def log_builder_information(self):
-    # TODO: align the key and value to make the build output prettier.
-    self.logger.info("Build information")
-    self.logger.info("=================")
-    self.logger.info("Profiles: {}".format(",".join(self.profile_dirs)))
-    self.logger.info("")
-    self.logger.info("Build variables")
-    self.logger.info("---------------")
-    for var, value in self.build_vars.items():
-      self.logger.info(f"{var} = {value}")
-
-    self.logger.info("")
-    self.logger.info("Environment variables")
-    self.logger.info("---------------------")
-    for var, value in self.env_vars.items():
-      self.logger.info(f"{var} = {value}")
-
-    self.logger.info("")
-    self.logger.info("Custom scripts")
-    self.logger.info("--------------")
-    for script in self.phase1_host_paths:
-      self.logger.info(f"phase1 host:   {script}")
-
-    for script in self.phase1_target_paths:
-      self.logger.info(f"phase1 target: {script}")
-
-    for script in self.phase2_host_paths:
-      self.logger.info(f"phase2 host:   {script}")
-
-    for script in self.phase2_target_paths:
-      self.logger.info(f"phase2 target: {script}")
-
   def start_session(self):
     if not os.path.isfile(self.session_file):
       with open(self.session_file, "w"):
@@ -206,7 +172,7 @@ def download_and_extract_image_if_necessary(self):
     if not os.path.isfile(self.cached_download_path):
       # Writing the code with wget gives better progress information than using
       # something like urllib3.
-      self.run_script_on_host([
+      self._run_script_on_host([
         "wget", "--progress=dot", "-e", "dotbytes=10M",
         "-O", self.cached_download_path, self.build_vars["image_url"],
       ])
@@ -217,22 +183,22 @@ def download_and_extract_image_if_necessary(self):
     self.logger.info(f"extracting {os.path.basename(self.cached_download_path)} into {self.output_filename}")
     # Writing the code as a shell with pv allows us to get a better progress
     # bar than implementing this code directly in Python.
-    self.run_script_on_host(f"{self.extract_image_path} {self.cached_download_path} | pv > {self.output_filename}", shell=True)
+    self._run_script_on_host(f"{self.extract_image_path} {self.cached_download_path} | pv > {self.output_filename}", shell=True)
 
   def setup_loop_device_and_mount_partitions(self):
     self.logger.info(f"expanding image to {self.build_vars['image_size']} with truncate")
-    self.run_script_on_host(["truncate", "-s", self.build_vars["image_size"], self.output_filename])
+    self._run_script_on_host(["truncate", "-s", self.build_vars["image_size"], self.output_filename])
 
     partition_end_in_mb = int(round(os.path.getsize(self.output_filename) / 1000.0 / 1000.0))
     partition_num = len(subprocess.check_output(["partx", "-g", self.output_filename]).decode("utf-8").strip().splitlines())
     self.logger.info(f"growing the last partition (partition number={partition_num}) to {partition_end_in_mb}MB")
 
-    self.run_script_on_host(["parted", self.output_filename, "resizepart", str(partition_num), str(partition_end_in_mb)])
+    self._run_script_on_host(["parted", self.output_filename, "resizepart", str(partition_num), str(partition_end_in_mb)])
 
     loop_device = subprocess.check_output(["losetup", "-P", "--show", "-f", self.output_filename]).decode("utf-8").strip()
 
-    self.run_script_on_host([self.loop_device_setup_path, loop_device])
-    self.cache_loop_device(loop_device)
+    self._run_script_on_host([self.loop_device_setup_path, loop_device])
+    self._cache_loop_device(loop_device)
 
   def prepare_chroot(self):
     loop_device = self.loop_device
@@ -244,7 +210,7 @@ def prepare_chroot(self):
       device_name = f"{loop_device}p{i}"
       mount_point = os.path.join(self.chroot_path, mount_point.lstrip("/"))
 
-      self.run_script_on_host(["mount", device_name, mount_point])
+      self._run_script_on_host(["mount", device_name, mount_point])
 
     self.logger.info("copy resolv.conf and qemu-user-static")
 
@@ -269,27 +235,27 @@ def prepare_chroot(self):
   def copy_files_to_chroot(self):
     for rootfs_path in self.rootfs_paths:
       # Use rsync instead of shutil.copytree as it is more easy to control permissions
-      self.run_script_on_host([
+      self._run_script_on_host([
         "rsync", "-r", "-og", "--chown", "root:root", "--stats",
         f"{rootfs_path}/",
         self.chroot_path,
       ])
 
   def run_phase1_host_scripts(self):
     for phase1_host_path in self.phase1_host_paths:
-      self.run_script_on_host(phase1_host_path)
+      self._run_script_on_host(phase1_host_path)
 
   def run_phase1_target_scripts(self):
     for phase1_target_path in self.phase1_target_paths:
-      self.run_script_on_target(phase1_target_path)
+      self._run_script_on_target(phase1_target_path)
 
   def run_phase2_host_scripts(self):
     for phase2_host_path in self.phase2_host_paths:
-      self.run_script_on_host(phase2_host_path)
+      self._run_script_on_host(phase2_host_path)
 
   def run_phase2_target_scripts(self):
     for phase2_target_path in self.phase2_target_paths:
-      self.run_script_on_target(phase2_target_path)
+      self._run_script_on_target(phase2_target_path)
 
   def cleanup_chroot(self):
     os.remove(os.path.join(self.chroot_path, "etc", "resolv.conf"))
@@ -303,22 +269,24 @@ def cleanup_chroot(self):
 
   def umount_everything(self):
     self.logger.info("Final system size:")
-    self.run_script_on_host(["df", "-h", self.chroot_path])
+    self._run_script_on_host(["df", "-h", self.chroot_path])
 
     self.logger.info("unmounting everything")
-    self.run_script_on_host(["umount", "-R", self.chroot_path])
-    self.run_script_on_host(["losetup", "-d", self.loop_device])
+    self._run_script_on_host(["umount", "-R", self.chroot_path])
+    self._run_script_on_host(["losetup", "-d", self.loop_device])
 
   def end_session(self):
     os.remove(self.session_file)
+    if os.path.exists(self.session_loop_device_file):
+      os.remove(self.session_loop_device_file)
 
   def run_step(self, f: Callable, always_run: bool = False):
     step = f.__name__
     extra_log = ""
     if not always_run:
-      if self.step_in_session(step):
+      if self._step_in_session(step):
         self.logger.info(f"skipped {step} as it already ran")
-        self.check_pause(step)
+        self._check_pause(step)
         return
     else:
       extra_log = "(idempotent step always run)" # To make it clear in the logs that idempotent steps always run
@@ -327,41 +295,75 @@ def run_step(self, f: Callable, always_run: bool = False):
     f()
 
     if not always_run:
-      self.record_step_in_session(step)
+      self._record_step_in_session(step)
+
+    self._check_pause(step)
 
-    self.check_pause(step)
+  def _parse_config(self, filename: str) -> tuple[dict, dict]:
+    config = configparser.ConfigParser()
+    # Preserve case sensitivity for configuration keys so that environment variables are properly exported.
+    config.optionxform = str
+    config.read(filename)
+    return (
+      dict(config["build"].items()),
+      dict(config["env"].items()),
+    )
+
+  def _log_builder_information(self):
+    # TODO: align the key and value to make the build output prettier.
+    self.logger.info("Build information")
+    self.logger.info("=================")
+    self.logger.info("Profiles: {}".format(",".join(self.profile_dirs)))
+    self.logger.info("")
+    self.logger.info("Build variables")
+    self.logger.info("---------------")
+    for var, value in self.build_vars.items():
+      self.logger.info(f"{var} = {value}")
+
+    self.logger.info("")
+    self.logger.info("Environment variables")
+    self.logger.info("---------------------")
+    for var, value in self.env_vars.items():
+      self.logger.info(f"{var} = {value}")
+
+    self.logger.info("")
+    self.logger.info("Custom scripts")
+    self.logger.info("--------------")
+    for script in self.phase1_host_paths:
+      self.logger.info(f"phase1 host:   {script}")
+
+    for script in self.phase1_target_paths:
+      self.logger.info(f"phase1 target: {script}")
+
+    for script in self.phase2_host_paths:
+      self.logger.info(f"phase2 host:   {script}")
+
+    for script in self.phase2_target_paths:
+      self.logger.info(f"phase2 target: {script}")
 
-  def check_pause(self, step: str):
+  def _check_pause(self, step: str):
     if self.build_vars.get("pause_after") == step:
       self.logger.warn(f"pausing after {step} as it is configured via the build var pause_after")
       print("Continue? [y/N] ", end="")
       if input().lower() != "y":
         raise SystemExit
 
-  def step_in_session(self, step: str) -> bool:
+  def _step_in_session(self, step: str) -> bool:
     with open(self.session_file) as f:
       return step in f.read()
 
-  def record_step_in_session(self, step: str):
+  def _record_step_in_session(self, step: str):
     with open(self.session_file, "a") as f:
       print(step, file=f)
 
-  def cache_loop_device(self, loop_device):
+  def _cache_loop_device(self, loop_device):
     with open(self.session_loop_device_file, "w") as f:
       f.write(loop_device)
 
     os.chmod(self.session_loop_device_file, 0o666)
     self._loop_device = loop_device
 
-  @property
-  def loop_device(self):
-    if getattr(self, "_loop_device", None) is None:
-      with open(self.session_loop_device_file) as f:
-        self._loop_device = f.read().strip()
-
-    return self._loop_device
-
-  def run_script_on_host(self, args: Sequence[str]|str, shell: bool = False):
+  def _run_script_on_host(self, args: Sequence[str]|str, shell: bool = False):
     self.logger.debug(f"running {args} with env {self.env_vars}")
     env_vars = os.environ.copy() # So PATH still works...
     env_vars.update(self.env_vars)
@@ -372,7 +374,7 @@ def run_script_on_host(self, args: Sequence[str]|str, shell: bool = False):
     else:
       subprocess.run(args, check=True, env=env_vars)
 
-  def run_script_on_target(self, script_path: str, args: Sequence[str] = []):
+  def _run_script_on_target(self, script_path: str, args: Sequence[str] = []):
     # Copy the script to inside the container
     script_filename = os.path.basename(script_path)
     script_path_in_target = os.path.join(self.chroot_path, "setup", script_filename)