Merge pull request 'api/machines: include disk and hw_config into machine details' (#2543) from hsjobeki/clan-core:hsjobeki-main into main
This commit is contained in:
@@ -184,7 +184,7 @@ def set_machine_disk_schema(
|
|||||||
raise ClanError(msg, description=f"Valid options: {ph.options}")
|
raise ClanError(msg, description=f"Valid options: {ph.options}")
|
||||||
|
|
||||||
header = f"""# ---
|
header = f"""# ---
|
||||||
# schema = "{schema_name}";
|
# schema = "{schema_name}"
|
||||||
# ---
|
# ---
|
||||||
# This file was automatically generated!
|
# This file was automatically generated!
|
||||||
# CHANGING this configuration requires wiping and reinstalling the machine
|
# CHANGING this configuration requires wiping and reinstalling the machine
|
||||||
|
|||||||
@@ -59,19 +59,12 @@ class Frontmatter:
|
|||||||
raise ValueError(msg)
|
raise ValueError(msg)
|
||||||
|
|
||||||
|
|
||||||
def extract_frontmatter(readme_content: str, err_scope: str) -> tuple[Frontmatter, str]:
|
def parse_frontmatter(readme_content: str) -> tuple[dict[str, Any] | None, str]:
|
||||||
"""
|
"""
|
||||||
Extracts TOML frontmatter from a README file content.
|
Extracts TOML frontmatter from a string
|
||||||
|
|
||||||
Parameters:
|
|
||||||
- readme_content (str): The content of the README file as a string.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
- str: The extracted frontmatter as a string.
|
|
||||||
- str: The content of the README file without the frontmatter.
|
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
- ValueError: If the README does not contain valid frontmatter.
|
- ClanError: If the toml frontmatter is invalid
|
||||||
"""
|
"""
|
||||||
# Pattern to match YAML frontmatter enclosed by triple-dashed lines
|
# Pattern to match YAML frontmatter enclosed by triple-dashed lines
|
||||||
frontmatter_pattern = r"^---\s+(.*?)\s+---\s?+(.*)$"
|
frontmatter_pattern = r"^---\s+(.*?)\s+---\s?+(.*)$"
|
||||||
@@ -92,11 +85,32 @@ def extract_frontmatter(readme_content: str, err_scope: str) -> tuple[Frontmatte
|
|||||||
msg = f"Error parsing TOML frontmatter: {e}"
|
msg = f"Error parsing TOML frontmatter: {e}"
|
||||||
raise ClanError(
|
raise ClanError(
|
||||||
msg,
|
msg,
|
||||||
description=f"Invalid TOML frontmatter. {err_scope}",
|
description="Invalid TOML frontmatter",
|
||||||
location="extract_frontmatter",
|
location="extract_frontmatter",
|
||||||
) from e
|
) from e
|
||||||
|
|
||||||
return Frontmatter(**frontmatter_parsed), remaining_content
|
return frontmatter_parsed, remaining_content
|
||||||
|
return None, readme_content
|
||||||
|
|
||||||
|
|
||||||
|
def extract_frontmatter(readme_content: str, err_scope: str) -> tuple[Frontmatter, str]:
|
||||||
|
"""
|
||||||
|
Extracts TOML frontmatter from a README file content.
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
- readme_content (str): The content of the README file as a string.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
- str: The extracted frontmatter as a string.
|
||||||
|
- str: The content of the README file without the frontmatter.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
- ValueError: If the README does not contain valid frontmatter.
|
||||||
|
"""
|
||||||
|
frontmatter_raw, remaining_content = parse_frontmatter(readme_content)
|
||||||
|
|
||||||
|
if frontmatter_raw:
|
||||||
|
return Frontmatter(**frontmatter_raw), remaining_content
|
||||||
|
|
||||||
# If no frontmatter is found, raise an error
|
# If no frontmatter is found, raise an error
|
||||||
msg = "Invalid README: Frontmatter not found."
|
msg = "Invalid README: Frontmatter not found."
|
||||||
|
|||||||
@@ -6,10 +6,13 @@ from pathlib import Path
|
|||||||
from typing import Literal
|
from typing import Literal
|
||||||
|
|
||||||
from clan_cli.api import API
|
from clan_cli.api import API
|
||||||
|
from clan_cli.api.modules import parse_frontmatter
|
||||||
from clan_cli.cmd import run_no_output
|
from clan_cli.cmd import run_no_output
|
||||||
from clan_cli.completions import add_dynamic_completer, complete_tags
|
from clan_cli.completions import add_dynamic_completer, complete_tags
|
||||||
|
from clan_cli.dirs import specific_machine_dir
|
||||||
from clan_cli.errors import ClanCmdError, ClanError
|
from clan_cli.errors import ClanCmdError, ClanError
|
||||||
from clan_cli.inventory import Machine, load_inventory_eval, set_inventory
|
from clan_cli.inventory import Machine, load_inventory_eval, set_inventory
|
||||||
|
from clan_cli.machines.hardware import HardwareConfig
|
||||||
from clan_cli.nix import nix_eval, nix_shell
|
from clan_cli.nix import nix_eval, nix_shell
|
||||||
from clan_cli.tags import list_nixos_machines_by_tags
|
from clan_cli.tags import list_nixos_machines_by_tags
|
||||||
|
|
||||||
@@ -34,32 +37,48 @@ def list_inventory_machines(flake_url: str | Path) -> dict[str, Machine]:
|
|||||||
@dataclass
|
@dataclass
|
||||||
class MachineDetails:
|
class MachineDetails:
|
||||||
machine: Machine
|
machine: Machine
|
||||||
has_hw_specs: bool = False
|
hw_config: HardwareConfig | None = None
|
||||||
# TODO:
|
disk_schema: str | None = None
|
||||||
# has_disk_specs: bool = False
|
|
||||||
|
|
||||||
|
import re
|
||||||
|
|
||||||
|
|
||||||
|
def extract_header(c: str) -> str:
|
||||||
|
header_lines = []
|
||||||
|
for line in c.splitlines():
|
||||||
|
match = re.match(r"^\s*#(.*)", line)
|
||||||
|
if match:
|
||||||
|
header_lines.append(match.group(1).strip())
|
||||||
|
else:
|
||||||
|
break # Stop once the header ends
|
||||||
|
return "\n".join(header_lines)
|
||||||
|
|
||||||
|
|
||||||
@API.register
|
@API.register
|
||||||
def get_inventory_machine_details(
|
def get_inventory_machine_details(flake_url: Path, machine_name: str) -> MachineDetails:
|
||||||
flake_url: str | Path, machine_name: str
|
|
||||||
) -> MachineDetails:
|
|
||||||
inventory = load_inventory_eval(flake_url)
|
inventory = load_inventory_eval(flake_url)
|
||||||
machine = inventory.machines.get(machine_name)
|
machine = inventory.machines.get(machine_name)
|
||||||
if machine is None:
|
if machine is None:
|
||||||
msg = f"Machine {machine_name} not found in inventory"
|
msg = f"Machine {machine_name} not found in inventory"
|
||||||
raise ClanError(msg)
|
raise ClanError(msg)
|
||||||
|
|
||||||
hw_config_path = (
|
hw_config = HardwareConfig.detect_type(flake_url, machine_name)
|
||||||
Path(flake_url) / "machines" / Path(machine_name) / "hardware-configuration.nix"
|
|
||||||
)
|
|
||||||
|
|
||||||
return MachineDetails(
|
machine_dir = specific_machine_dir(flake_url, machine_name)
|
||||||
machine=machine,
|
disk_schema: str | None = None
|
||||||
has_hw_specs=hw_config_path.exists(),
|
disk_path = machine_dir / "disko.nix"
|
||||||
)
|
if disk_path.exists():
|
||||||
|
with disk_path.open() as f:
|
||||||
|
content = f.read()
|
||||||
|
header = extract_header(content)
|
||||||
|
data, _rest = parse_frontmatter(header)
|
||||||
|
if data:
|
||||||
|
disk_schema = data.get("schema")
|
||||||
|
|
||||||
|
return MachineDetails(machine=machine, hw_config=hw_config, disk_schema=disk_schema)
|
||||||
|
|
||||||
|
|
||||||
@API.register
|
|
||||||
def list_nixos_machines(flake_url: str | Path) -> list[str]:
|
def list_nixos_machines(flake_url: str | Path) -> list[str]:
|
||||||
cmd = nix_eval(
|
cmd = nix_eval(
|
||||||
[
|
[
|
||||||
|
|||||||
Reference in New Issue
Block a user