clan-cli: Init clan machines import

pkgs/clan-cli/clan_cli/clan_uri.py

@@ -4,8 +4,6 @@ import urllib.request
 from dataclasses import dataclass
 from pathlib import Path
 
-from .errors import ClanError
-
 
 @dataclass
 class FlakeId:
@@ -54,47 +52,21 @@ class FlakeId:
         return not self.is_local()
 
 
+def _parse_url(comps: urllib.parse.ParseResult) -> FlakeId:
+    if comps.scheme == "" or "file" in comps.scheme:
+        res_p = Path(comps.path).expanduser().resolve()
+        flake_id = FlakeId(str(res_p))
+    else:
+        flake_id = FlakeId(comps.geturl())
+    return flake_id
+
+
 # Define the ClanURI class
 @dataclass
 class ClanURI:
     flake: FlakeId
     machine_name: str
 
-    # Initialize the class with a clan:// URI
-    def __init__(self, uri: str) -> None:
-        # users might copy whitespace along with the uri
-        uri = uri.strip()
-        self._orig_uri = uri
-
-        # Check if the URI starts with clan://
-        # If it does, remove the clan:// prefix
-        if uri.startswith("clan://"):
-            nested_uri = uri[7:]
-        else:
-            msg = f"Invalid uri: expected clan://, got {uri}"
-            raise ClanError(msg)
-
-        # Parse the URI into components
-        # url://netloc/path;parameters?query#fragment
-        components: urllib.parse.ParseResult = urllib.parse.urlparse(nested_uri)
-
-        # Replace the query string in the components with the new query string
-        clean_comps = components._replace(query=components.query, fragment="")
-
-        # Parse the URL into a ClanUrl object
-        self.flake = self._parse_url(clean_comps)
-        self.machine_name = "defaultVM"
-        if components.fragment:
-            self.machine_name = components.fragment
-
-    def _parse_url(self, comps: urllib.parse.ParseResult) -> FlakeId:
-        if comps.scheme == "" or "file" in comps.scheme:
-            res_p = Path(comps.path).expanduser().resolve()
-            flake_id = FlakeId(str(res_p))
-        else:
-            flake_id = FlakeId(comps.geturl())
-        return flake_id
-
     def get_url(self) -> str:
         return str(self.flake)
 
@@ -104,13 +76,31 @@ class ClanURI:
         url: str,
         machine_name: str | None = None,
     ) -> "ClanURI":
-        clan_uri = ""
+        uri = url
-        if not url.startswith("clan://"):
-            clan_uri += "clan://"
-
-        clan_uri += url
 
         if machine_name:
-            clan_uri += f"#{machine_name}"
+            uri += f"#{machine_name}"
 
-        return cls(clan_uri)
+        # users might copy whitespace along with the uri
+        uri = uri.strip()
+
+        # Check if the URI starts with clan://
+        # If it does, remove the clan:// prefix
+        prefix = "clan://"
+        if uri.startswith(prefix):
+            uri = uri[len(prefix) :]
+
+        # Parse the URI into components
+        # url://netloc/path;parameters?query#fragment
+        components: urllib.parse.ParseResult = urllib.parse.urlparse(uri)
+
+        # Replace the query string in the components with the new query string
+        clean_comps = components._replace(query=components.query, fragment="")
+
+        # Parse the URL into a ClanUrl object
+        flake = _parse_url(clean_comps)
+        machine_name = "defaultVM"
+        if components.fragment:
+            machine_name = components.fragment
+
+        return cls(flake, machine_name)

pkgs/clan-cli/clan_cli/flash/cli.py

@@ -1,7 +1,7 @@
 # !/usr/bin/env python3
 import argparse
 
-from .flash_command import register_flash_apply_parser
+from .flash_cmd import register_flash_apply_parser
 from .list import register_flash_list_parser
 
 
@@ -17,13 +17,8 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
     apply_parser = subparser.add_parser(
         "apply",
         help="Flash a machine",
-        formatter_class=argparse.RawTextHelpFormatter,
     )
     register_flash_apply_parser(apply_parser)
 
-    list_parser = subparser.add_parser(
+    list_parser = subparser.add_parser("list", help="List options")
-        "list",
-        help="List options",
-        formatter_class=argparse.RawTextHelpFormatter,
-    )
     register_flash_list_parser(list_parser)

pkgs/clan-cli/clan_cli/history/add.py

@@ -111,7 +111,9 @@ def add_history_command(args: argparse.Namespace) -> None:
 
 # takes a (sub)parser and configures it
 def register_add_parser(parser: argparse.ArgumentParser) -> None:
-    parser.add_argument("uri", type=ClanURI, help="Path to the flake", default=".")
+    parser.add_argument(
+        "uri", type=ClanURI.from_str, help="Path to the flake", default="."
+    )
     parser.add_argument(
         "--all", help="Add all machines", default=False, action="store_true"
     )

pkgs/clan-cli/clan_cli/machines/cli.py

@@ -4,6 +4,7 @@ import argparse
 from .create import register_create_parser
 from .delete import register_delete_parser
 from .hardware import register_hw_generate
+from .import_cmd import register_import_parser
 from .install import register_install_parser
 from .list import register_list_parser
 from .update import register_update_parser
@@ -113,3 +114,22 @@ For more detailed information, visit: https://docs.clan.lol/getting-started/depl
         formatter_class=argparse.RawTextHelpFormatter,
     )
     register_install_parser(install_parser)
+
+    import_parser = subparser.add_parser(
+        "import",
+        help="Import a machine",
+        description="Import a template machine from a local or remote source.",
+        epilog=(
+            """
+
+Examples:
+    $ clan machines import flash-installer
+    Will import a machine from the flash-installer template from clan-core.
+
+    $ clan machines import flash-installer --src https://git.clan.lol/clan/clan-core
+    Will import a machine from the flash-installer template from clan-core but with the source set to the provided URL.
+    """
+        ),
+        formatter_class=argparse.RawTextHelpFormatter,
+    )
+    register_import_parser(import_parser)

pkgs/clan-cli/clan_cli/machines/import_cmd.py

@@ -0,0 +1,160 @@
+import argparse
+import logging
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from clan_cli.api import API
+from clan_cli.clan.create import git_command
+from clan_cli.clan_uri import FlakeId
+from clan_cli.cmd import Log, run
+from clan_cli.dirs import clan_templates, get_clan_flake_toplevel_or_env
+from clan_cli.errors import ClanError
+from clan_cli.machines.list import list_nixos_machines
+from clan_cli.machines.machines import Machine
+from clan_cli.nix import nix_command
+
+log = logging.getLogger(__name__)
+
+
+def validate_directory(root_dir: Path, machine_name: str) -> None:
+    machines_dir = root_dir / "machines" / machine_name
+    for root, _, files in root_dir.walk():
+        for file in files:
+            file_path = Path(root) / file
+            if not file_path.is_relative_to(machines_dir):
+                msg = f"File {file_path} is not in the 'machines/{machine_name}' directory."
+                description = "Template machines are only allowed to contain files in the 'machines/{machine_name}' directory."
+                raise ClanError(msg, description=description)
+
+
+@dataclass
+class ImportOptions:
+    target: FlakeId
+    src: Machine
+    rename: str | None = None
+
+
+@API.register
+def import_machine(opts: ImportOptions) -> None:
+    if not opts.target.is_local():
+        msg = f"Clan {opts.target} is not a local clan."
+        description = "Import machine only works on local clans"
+        raise ClanError(msg, description=description)
+    clan_dir = opts.target.path
+
+    log.debug(f"Importing machine '{opts.src.name}' from {opts.src.flake}")
+
+    if opts.src.name == opts.rename:
+        msg = "Rename option must be different from the template machine name"
+        raise ClanError(msg)
+
+    if opts.src.name in list_nixos_machines(clan_dir) and not opts.rename:
+        msg = f"{opts.src.name} is already defined in {clan_dir}"
+        description = (
+            "Please add the --rename option to import the machine with a different name"
+        )
+        raise ClanError(msg, description=description)
+
+    machine_name = opts.src.name if not opts.rename else opts.rename
+    dst = clan_dir / "machines" / machine_name
+
+    if dst.exists():
+        msg = f"Machine {machine_name} already exists in {clan_dir}"
+        description = (
+            "Please delete the existing machine or import with a different name"
+        )
+        raise ClanError(msg, description=description)
+
+    with TemporaryDirectory() as tmpdir:
+        tmpdirp = Path(tmpdir)
+        command = nix_command(
+            [
+                "flake",
+                "init",
+                "-t",
+                opts.src.get_id(),
+            ]
+        )
+        run(command, log=Log.NONE, cwd=tmpdirp)
+
+        validate_directory(tmpdirp, opts.src.name)
+        src = tmpdirp / "machines" / opts.src.name
+
+        if (
+            not (src / "configuration.nix").exists()
+            and not (src / "inventory.json").exists()
+        ):
+            msg = f"Template machine {opts.src.name} does not contain a configuration.nix or inventory.json"
+            description = (
+                "Template machine must contain a configuration.nix or inventory.json"
+            )
+            raise ClanError(msg, description=description)
+
+        def log_copy(src: str, dst: str) -> None:
+            relative_dst = dst.replace(f"{clan_dir}/", "")
+            log.info(f"Add file: {relative_dst}")
+            shutil.copy2(src, dst)
+
+        shutil.copytree(src, dst, ignore_dangling_symlinks=True, copy_function=log_copy)
+
+    run(git_command(clan_dir, "add", f"machines/{machine_name}"), cwd=clan_dir)
+
+    if (dst / "inventory.json").exists():
+        # TODO: Implement inventory import
+        msg = "Inventory import not implemented yet"
+        raise NotImplementedError(msg)
+        # inventory = load_inventory_json(clan_dir)
+
+        # inventory.machines[machine_name] = Inventory_Machine(
+        #     name=machine_name,
+        #     deploy=MachineDeploy(targetHost=None),
+        # )
+        # set_inventory(inventory, clan_dir, "Imported machine from template")
+
+
+def import_command(args: argparse.Namespace) -> None:
+    if args.flake:
+        target = args.flake
+    else:
+        tmp = get_clan_flake_toplevel_or_env()
+        target = FlakeId(str(tmp)) if tmp else None
+
+    if not target:
+        msg = "No clan found."
+        description = (
+            "Run this command in a clan directory or specify the --flake option"
+        )
+        raise ClanError(msg, description=description)
+
+    src_uri = args.src
+    if not src_uri:
+        src_uri = FlakeId(str(clan_templates()))
+
+    opts = ImportOptions(
+        target=target,
+        src=Machine(flake=src_uri, name=args.machine_name),
+        rename=args.rename,
+    )
+    import_machine(opts)
+
+
+def register_import_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "machine_name",
+        type=str,
+        help="The name of the machine to import",
+    )
+    parser.add_argument(
+        "--src",
+        type=FlakeId,
+        help="The source flake to import the machine from",
+    )
+    parser.add_argument(
+        "--rename",
+        type=str,
+        help="Rename the imported machine",
+    )
+
+    parser.set_defaults(func=import_command)

pkgs/clan-cli/tests/test_clan_uri.py

@@ -5,46 +5,46 @@ from clan_cli.clan_uri import ClanURI
 
 def test_get_url() -> None:
     # Create a ClanURI object from a remote URI with parameters
-    uri = ClanURI("clan://https://example.com?password=1234#myVM")
+    uri = ClanURI.from_str("clan://https://example.com?password=1234#myVM")
     assert uri.get_url() == "https://example.com?password=1234"
 
-    uri = ClanURI("clan://~/Downloads")
+    uri = ClanURI.from_str("clan://~/Downloads")
     assert uri.get_url().endswith("/Downloads")
 
-    uri = ClanURI("clan:///home/user/Downloads")
+    uri = ClanURI.from_str("clan:///home/user/Downloads")
     assert uri.get_url() == "/home/user/Downloads"
 
-    uri = ClanURI("clan://file:///home/user/Downloads")
+    uri = ClanURI.from_str("clan://file:///home/user/Downloads")
     assert uri.get_url() == "/home/user/Downloads"
 
 
 def test_local_uri() -> None:
     # Create a ClanURI object from a local URI
-    uri = ClanURI("clan://file:///home/user/Downloads")
+    uri = ClanURI.from_str("clan://file:///home/user/Downloads")
     assert uri.flake.path == Path("/home/user/Downloads")
 
 
 def test_is_remote() -> None:
     # Create a ClanURI object from a remote URI
-    uri = ClanURI("clan://https://example.com")
+    uri = ClanURI.from_str("clan://https://example.com")
     assert uri.flake.url == "https://example.com"
 
 
 def test_direct_local_path() -> None:
     # Create a ClanURI object from a remote URI
-    uri = ClanURI("clan://~/Downloads")
+    uri = ClanURI.from_str("clan://~/Downloads")
     assert uri.get_url().endswith("/Downloads")
 
 
 def test_direct_local_path2() -> None:
     # Create a ClanURI object from a remote URI
-    uri = ClanURI("clan:///home/user/Downloads")
+    uri = ClanURI.from_str("clan:///home/user/Downloads")
     assert uri.get_url() == "/home/user/Downloads"
 
 
 def test_remote_with_clanparams() -> None:
     # Create a ClanURI object from a remote URI with parameters
-    uri = ClanURI("clan://https://example.com")
+    uri = ClanURI.from_str("clan://https://example.com")
 
     assert uri.machine_name == "defaultVM"
     assert uri.flake.url == "https://example.com"

pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py

@@ -65,7 +65,7 @@ class ClanStore:
         self._emitter.connect(signal, cb)
 
     def set_logging_vm(self, ident: str) -> VMObject | None:
-        vm = self.get_vm(ClanURI(f"clan://{ident}"))
+        vm = self.get_vm(ClanURI.from_str(f"clan://{ident}"))
 
         if vm is not None:
             self._logging_vm = vm

pkgs/clan-vm-manager/clan_vm_manager/views/list.py

@@ -335,7 +335,7 @@ class ClanList(Gtk.Box):
 
     def on_join_request(self, source: Any, url: str) -> None:
         log.debug("Join request: %s", url)
-        clan_uri = ClanURI(url)
+        clan_uri = ClanURI.from_str(url)
         JoinList.use().push(clan_uri, self.on_after_join)
 
     def on_after_join(self, source: JoinValue) -> None:

templates/flake.nix

@@ -15,6 +15,10 @@
           description = "Flake-parts";
           path = ./flake-parts;
         };
+        flash-installer = {
+          description = "Flash installer";
+          path = ./flash-installer;
+        };
       };
     };
 }

templates/flash-installer/machines/flash-installer/configuration.nix

@@ -0,0 +1,7 @@
+{
+  imports = [
+
+  ];
+
+  # Hello World
+}