feat: implement existing partition install mode in TUI, GUI, and backend

Author: iluxz

archiso-mochios/airootfs/usr/lib/mochiinstall/installer.py

@@ -262,6 +262,24 @@ def detect_disk_partitions_summary(disk):
     return "\n".join(lines)
 
 
+def detect_existing_parts(disk):
+    """detect EFI, swap, and data partitions on a disk for existing mode.
+    returns dict with 'efi', 'swap', 'data' keys (partition paths or None)."""
+    parts = detect_partitions(disk)
+    result = {"efi": None, "swap": None, "data": None}
+    for p in parts:
+        fstype = p.get("fstype", "")
+        label = p.get("label", "").upper()
+        name = p["name"]
+        if fstype == "vfat" and (label == "EFI" or "efi" in name.lower()):
+            result["efi"] = name
+        elif fstype == "swap" or label == "SWAP":
+            result["swap"] = name
+        elif fstype in ("ext4", "btrfs", "xfs") and result["data"] is None:
+            result["data"] = name
+    return result
+
+
 def select_disk(disk):
     if not disk or not disk.startswith("/dev/"):
         raise ValueError(f"invalid disk: {disk}")
@@ -441,25 +459,22 @@ def ch(cmd, to=120, check=True, input_data=None):
     ch(["mkinitcpio", "-P"], to=300)
 
     lfn("writing fstab...")
+    fstab_lines = []
+    if efi_uuid:
+        fstab_lines.append(f"UUID={efi_uuid} /boot vfat rw,noatime,fmask=0022,dmask=0022 0 2")
+    if swap_uuid:
+        fstab_lines.append(f"UUID={swap_uuid} swap swap defaults 0 0")
     if filesystem == "btrfs":
-        fstab = (
-            f"UUID={efi_uuid} /boot vfat rw,noatime,fmask=0022,dmask=0022 0 2\n"
-            f"UUID={swap_uuid} swap swap defaults 0 0\n"
-            f"UUID={root_uuid} /mnt/btrfs btrfs rw,noatime,compress=zstd,subvolid=5 0 0\n"
-            f"UUID={root_uuid} / btrfs rw,noatime,compress=zstd,subvol=root_a 0 0\n"
-            f"UUID={root_uuid} /home btrfs rw,noatime,compress=zstd,subvol=home 0 0\n"
-            f"UUID={root_uuid} /var btrfs rw,noatime,compress=zstd,subvol=var 0 0\n"
-            f"UUID={root_uuid} /var/cache btrfs rw,noatime,compress=zstd,subvol=cache 0 0\n"
-        )
+        fstab_lines.append(f"UUID={root_uuid} /mnt/btrfs btrfs rw,noatime,compress=zstd,subvolid=5 0 0")
+        fstab_lines.append(f"UUID={root_uuid} / btrfs rw,noatime,compress=zstd,subvol=root_a 0 0")
+        fstab_lines.append(f"UUID={root_uuid} /home btrfs rw,noatime,compress=zstd,subvol=home 0 0")
+        fstab_lines.append(f"UUID={root_uuid} /var btrfs rw,noatime,compress=zstd,subvol=var 0 0")
+        fstab_lines.append(f"UUID={root_uuid} /var/cache btrfs rw,noatime,compress=zstd,subvol=cache 0 0")
         os.makedirs(f"{target}/mnt/btrfs", exist_ok=True)
     else:
-        fstab = (
-            f"UUID={efi_uuid} /boot vfat rw,noatime,fmask=0022,dmask=0022 0 2\n"
-            f"UUID={swap_uuid} swap swap defaults 0 0\n"
-            f"UUID={root_uuid} / ext4 rw,noatime 0 1\n"
-            f"UUID={root_uuid} /home ext4 rw,noatime 0 2\n"
-        )
-    (t / "etc/fstab").write_text(fstab)
+        fstab_lines.append(f"UUID={root_uuid} / ext4 rw,noatime 0 1")
+        fstab_lines.append(f"UUID={root_uuid} /home ext4 rw,noatime 0 2")
+    (t / "etc/fstab").write_text("\n".join(fstab_lines) + "\n")
     ch(["systemctl", "daemon-reload"])
 
     if filesystem == "btrfs":
@@ -735,40 +750,94 @@ def do_install(target="/mnt/mochios", config=None, log_fn=None, abort_flag=None)
         _unmount_all()
 
         disk = select_disk(config.get("disk"))
-        bios_part, efi_part, swap_part, root_part_p = partition_disk(disk, log_fn)
-        mounts_cleanup_needed = True
+        disk_mode = config.get("disk_mode", "wipe")
 
-        if abort_flag and abort_flag():
-            raise RuntimeError("installation aborted")
-        format_efi(efi_part, log_fn)
-        format_swap(swap_part, log_fn)
+        if disk_mode == "existing":
+            existing = detect_existing_parts(disk)
+            efi_part = existing["efi"]
+            swap_part = existing["swap"]
+            root_part_p = config.get("existing_root") or existing["data"]
 
-        filesystem = config.get("filesystem", "btrfs")
-        if filesystem == "ext4":
-            format_ext4(root_part_p, log_fn)
-        else:
-            format_btrfs(root_part_p, log_fn)
+            if not root_part_p:
+                raise RuntimeError("no suitable root partition found on disk")
 
-        if abort_flag and abort_flag():
-            raise RuntimeError("installation aborted")
-        _CLEANUP_TARGET[0] = target
-        if filesystem == "ext4":
-            setup_ext4(root_part_p, target, log_fn)
+            log_fn(f"using existing partitions on {disk}:")
+            log_fn(f"  root: {root_part_p}")
+            if efi_part:
+                log_fn(f"  efi:  {efi_part}")
+            else:
+                log_fn("  [yellow]no EFI partition found — you may need to set up boot manually[/]")
+            if swap_part:
+                log_fn(f"  swap: {swap_part}")
+            else:
+                log_fn("  [yellow]no swap partition found — no swap will be active[/]")
+
+            mounts_cleanup_needed = True
+
+            if abort_flag and abort_flag():
+                raise RuntimeError("installation aborted")
+
+            filesystem = config.get("filesystem", "btrfs")
+            if filesystem == "ext4":
+                format_ext4(root_part_p, log_fn)
+            else:
+                format_btrfs(root_part_p, log_fn)
+
+            _CLEANUP_TARGET[0] = target
+            if filesystem == "ext4":
+                setup_ext4(root_part_p, target, log_fn)
+            else:
+                setup_btrfs(root_part_p, target, log_fn)
+
+            root_uuid = get_uuid(root_part_p)
+            efi_uuid = get_uuid(efi_part) if efi_part else ""
+            swap_uuid = get_uuid(swap_part) if swap_part else ""
+
+            if filesystem == "btrfs":
+                log_fn("configuring snapper for root...")
+                run(["arch-chroot", target, "snapper", "-c", "root", "create-config", "/"], check=False, timeout=30)
+                run(["arch-chroot", target, "systemctl", "enable", "snapper-timeline.timer"], check=False)
+                run(["arch-chroot", target, "systemctl", "enable", "snapper-cleanup.timer"], check=False)
+
+            if swap_part:
+                run(["swapon", swap_part], check=False)
+            if efi_part:
+                run(["mount", efi_part, f"{target}/boot"])
         else:
-            setup_btrfs(root_part_p, target, log_fn)
+            bios_part, efi_part, swap_part, root_part_p = partition_disk(disk, log_fn)
+            mounts_cleanup_needed = True
+
+            if abort_flag and abort_flag():
+                raise RuntimeError("installation aborted")
+            format_efi(efi_part, log_fn)
+            format_swap(swap_part, log_fn)
+
+            filesystem = config.get("filesystem", "btrfs")
+            if filesystem == "ext4":
+                format_ext4(root_part_p, log_fn)
+            else:
+                format_btrfs(root_part_p, log_fn)
+
+            if abort_flag and abort_flag():
+                raise RuntimeError("installation aborted")
+            _CLEANUP_TARGET[0] = target
+            if filesystem == "ext4":
+                setup_ext4(root_part_p, target, log_fn)
+            else:
+                setup_btrfs(root_part_p, target, log_fn)
 
-        efi_uuid = get_uuid(efi_part)
-        swap_uuid = get_uuid(swap_part)
-        root_uuid = get_uuid(root_part_p)
+            efi_uuid = get_uuid(efi_part)
+            swap_uuid = get_uuid(swap_part)
+            root_uuid = get_uuid(root_part_p)
 
-        if filesystem == "btrfs":
-            log_fn("configuring snapper for root...")
-            run(["arch-chroot", target, "snapper", "-c", "root", "create-config", "/"], check=False, timeout=30)
-            run(["arch-chroot", target, "systemctl", "enable", "snapper-timeline.timer"], check=False)
-            run(["arch-chroot", target, "systemctl", "enable", "snapper-cleanup.timer"], check=False)
+            if filesystem == "btrfs":
+                log_fn("configuring snapper for root...")
+                run(["arch-chroot", target, "snapper", "-c", "root", "create-config", "/"], check=False, timeout=30)
+                run(["arch-chroot", target, "systemctl", "enable", "snapper-timeline.timer"], check=False)
+                run(["arch-chroot", target, "systemctl", "enable", "snapper-cleanup.timer"], check=False)
 
-        run(["swapon", swap_part], check=False)
-        run(["mount", efi_part, f"{target}/boot"])
+            run(["swapon", swap_part], check=False)
+            run(["mount", efi_part, f"{target}/boot"])
 
         extra_all = config.get("extra_pkgs", []) or []
         repo_extra = [p for p in extra_all if isinstance(p, str) and p not in CUSTOM_EXTRAS]

archiso-mochios/airootfs/usr/lib/mochiinstall/mochiinstall_app.py

@@ -13,7 +13,7 @@
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 from mochi_ascii import MOCHI_ASCII
-from installer import do_install, set_progress_cb, detect_partitions, detect_disk_partitions_summary
+from installer import do_install, set_progress_cb, detect_partitions, detect_disk_partitions_summary, detect_existing_parts
 
 
 STEP_NAMES = ["Hostname", "Disk", "Partitions", "Filesystem", "Kernel", "Extras", "Desktop & Boot", "User", "Review"]
@@ -161,6 +161,12 @@ def _push_next(self) -> bool:
             hint = self.query_one("#disk_hint", Label) if self.body.query("#disk_hint") else None
             if hint: hint.update("[red]select a disk first![/]")
             return False
+        if self.step == 2:
+            mode = self.config.get("disk_mode", "wipe")
+            if mode == "existing" and not self.config.get("existing_root"):
+                hint = self.query_one("#part_hint", Label) if self.body.query("#part_hint") else None
+                if hint: hint.update("[red]select a partition or choose wipe![/]")
+                return False
         if self.step == 7:
             pw = self.config.get("password", "")
             pw_c = self.config.get("password_confirm", "")
@@ -208,6 +214,11 @@ def on_selection_list_selection_toggled(self, event: SelectionList.SelectionTogg
             hint = self.query_one("#disk_hint", Label)
             hint.update("")
             self._nav(can_next=has_disk, next_label="next")
+        elif self.step == 2:
+            sel = self.query_one("#disk_mode", SelectionList)
+            if sel.selected:
+                self.config["disk_mode"] = sel.selected[0]
+            self.render_step()
 
     def on_input_submitted(self, event: Input.Submitted) -> None:
         if self.step == 0:
@@ -296,14 +307,36 @@ def step_disk(self) -> None:
         self.body.mount(self._wrap(*kids))
 
     def step_partition(self) -> None:
-        """show partition info for selected disk."""
+        """show partition info and choose wipe vs existing."""
         disk = self.config.get("disk", "")
         kids = [Static(f"[bold]partitions on {disk}[/]")]
 
         part_summary = detect_disk_partitions_summary(disk)
         kids.append(Static(f"[dim]{part_summary}[/]"))
         kids.append(Static(""))
-        kids.append(Static("[dim]full 'use existing partition' mode coming soon — currently wipes the whole disk[/]"))
+
+        mode = self.config.get("disk_mode", "wipe")
+        kids.append(Label("", id="part_hint"))
+        kids.append(SelectionList(
+            ("wipe entire disk (delete all partitions, create new mochios layout)", "wipe", mode == "wipe"),
+            ("use existing partition (install mochios alongside other OS)", "existing", mode == "existing"),
+            id="disk_mode",
+        ))
+
+        if mode == "existing":
+            existing = detect_existing_parts(disk)
+            linux_parts = []
+            parts = detect_partitions(disk)
+            for p in parts:
+                if p["fstype"] in ("ext4", "btrfs", "xfs", ""):
+                    label = f"{p['name']}  {p['size']}  {p['fstype'] or 'unknown'}"
+                    linux_parts.append((label, p["name"], p["name"] == self.config.get("existing_root", "")))
+            if linux_parts:
+                kids.append(Static(""))
+                kids.append(Static("[bold]select root partition for mochios:[/]"))
+                kids.append(SelectionList(*linux_parts, id="existing_root"))
+            else:
+                kids.append(Static("[yellow]no suitable partitions found — create one first or choose wipe[/]"))
 
         self._nav()
         self.body.mount(self._wrap(*kids))
@@ -394,11 +427,14 @@ def step_summary(self) -> None:
         disk_set = bool(c["disk"])
         extras = ", ".join(c.get("extra_pkgs", [])) or "none"
         kernels = ", ".join(c.get("kernels", ["linux"])) or "linux"
+        disk_mode = c.get("disk_mode", "wipe")
+        mode_label = "wipe entire disk" if disk_mode == "wipe" else f"use partition {c.get('existing_root', '?')}"
         self.body.mount(self._wrap(
             Static("[bold underline]review configuration[/]"),
             Static(""),
             Static(f"  hostname:    [white]{c['hostname']}[/]"),
-            Static(f"  disk:        [red]{c['disk'] or 'NOT SELECTED'}[/] {'[red bold](required!)' if not disk_set else '[dim](will be wiped)[/]'}"),
+            Static(f"  disk:        [red]{c['disk'] or 'NOT SELECTED'}[/] {'[red bold](required!)' if not disk_set else ''}"),
+            Static(f"  disk mode:   [white]{mode_label}[/]"),
             Static(f"  filesystem:  [white]{c.get('filesystem', 'btrfs')}[/]"),
             Static(f"  desktop:     [white]{c['de']}[/]"),
             Static(f"  greeter:     [white]{c.get('greeter', 'sddm')}[/]"),
@@ -407,7 +443,7 @@ def step_summary(self) -> None:
             Static(f"  extras:      [white]{extras}[/]"),
             Static(f"  user:        [white]{c['username']}[/]"),
             Static(""),
-            Static("[red bold]WARNING: this will erase all data on the selected disk" if disk_set else "[red bold]select a disk first!"),
+            Static("[red bold]WARNING: this will erase data on the selected disk/partition" if disk_set else "[red bold]select a disk first!"),
         ))
         self._nav(next_label="install!" if disk_set else "", can_next=disk_set)
 
@@ -420,6 +456,17 @@ def collect_current(self) -> None:
                 sel = self.query_one("#disk_list", SelectionList)
                 if sel.selected:
                     self.config["disk"] = sel.selected[0]
+            elif self.step == 2:
+                sel = self.query_one("#disk_mode", SelectionList)
+                if sel.selected:
+                    self.config["disk_mode"] = sel.selected[0]
+                if self.config.get("disk_mode") == "existing":
+                    try:
+                        root_sel = self.query_one("#existing_root", SelectionList)
+                        if root_sel.selected:
+                            self.config["existing_root"] = root_sel.selected[0]
+                    except Exception:
+                        pass
             elif self.step == 3:
                 sel = self.query_one("#fs", SelectionList)
                 if sel.selected: