API: init methods: hw_generate, dns discovery

pkgs/clan-cli/clan_cli/__init__.py

@@ -5,11 +5,11 @@ from pathlib import Path
 from types import ModuleType
 
 # These imports are unused, but necessary for @API.register to run once.
-from clan_cli.api import directory
+from clan_cli.api import directory, mdns_discovery
 from clan_cli.arg_actions import AppendOptionAction
 from clan_cli.clan import show
 
-__all__ = ["directory"]
+__all__ = ["directory", "mdns_discovery"]
 
 from . import (
     backups,

pkgs/clan-cli/clan_cli/api/mdns_discovery.py

@@ -0,0 +1,116 @@
+import argparse
+import re
+from dataclasses import dataclass
+
+from clan_cli.cmd import run_no_stdout
+from clan_cli.nix import nix_shell
+
+from . import API
+
+
+@dataclass
+class Host:
+    # Part of the discovery
+    interface: str
+    protocol: str
+    name: str
+    type_: str
+    domain: str
+    # Optional, only if more data is available
+    host: str | None
+    ip: str | None
+    port: str | None
+    txt: str | None
+
+
+@dataclass
+class DNSInfo:
+    """ "
+    mDNS/DNS-SD services discovered on the network
+    """
+
+    services: dict[str, Host]
+
+
+def decode_escapes(s: str) -> str:
+    return re.sub(r"\\(\d{3})", lambda x: chr(int(x.group(1))), s)
+
+
+def parse_avahi_output(output: str) -> DNSInfo:
+    dns_info = DNSInfo(services={})
+    for line in output.splitlines():
+        parts = line.split(";")
+        # New service discovered
+        # print(parts)
+        if parts[0] == "+" and len(parts) >= 6:
+            interface, protocol, name, type_, domain = parts[1:6]
+
+            name = decode_escapes(name)
+
+            dns_info.services[name] = Host(
+                interface=interface,
+                protocol=protocol,
+                name=name,
+                type_=decode_escapes(type_),
+                domain=domain,
+                host=None,
+                ip=None,
+                port=None,
+                txt=None,
+            )
+
+        # Resolved more data for already discovered services
+        elif parts[0] == "=" and len(parts) >= 9:
+            interface, protocol, name, type_, domain, host, ip, port = parts[1:9]
+
+            name = decode_escapes(name)
+
+            if name in dns_info.services:
+                dns_info.services[name].host = decode_escapes(host)
+                dns_info.services[name].ip = ip
+                dns_info.services[name].port = port
+                if len(parts) > 9:
+                    dns_info.services[name].txt = decode_escapes(parts[9])
+            else:
+                dns_info.services[name] = Host(
+                    interface=parts[1],
+                    protocol=parts[2],
+                    name=name,
+                    type_=decode_escapes(parts[4]),
+                    domain=parts[5],
+                    host=decode_escapes(parts[6]),
+                    ip=parts[7],
+                    port=parts[8],
+                    txt=decode_escapes(parts[9]) if len(parts) > 9 else None,
+                )
+
+    return dns_info
+
+
+@API.register
+def show_mdns() -> DNSInfo:
+    cmd = nix_shell(
+        ["nixpkgs#avahi"],
+        [
+            "avahi-browse",
+            "--all",
+            "--resolve",
+            "--parsable",
+            "-l",  # Ignore local services
+            "--terminate",
+        ],
+    )
+    proc = run_no_stdout(cmd)
+    data = parse_avahi_output(proc.stdout)
+
+    return data
+
+
+def mdns_command(args: argparse.Namespace) -> None:
+    dns_info = show_mdns()
+    for name, info in dns_info.services.items():
+        print(f"Hostname: {name} - ip: {info.ip}")
+
+
+def register_mdns(parser: argparse.ArgumentParser) -> None:
+    parser.set_defaults(func=mdns_command)

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

@@ -20,6 +20,72 @@ class HardwareInfo:
     system: str | None
 
 
+@API.register
+def show_machine_hardware_info(
+    clan_dir: str | Path, machine_name: str
+) -> HardwareInfo | None:
+    """
+    Show hardware information for a machine returns None if none exist.
+    """
+
+    hw_file = Path(f"{clan_dir}/machines/{machine_name}/hardware-configuration.nix")
+
+    is_template = hw_file.exists() and "throw" in hw_file.read_text()
+    if not hw_file.exists() or is_template:
+        return None
+
+    system = show_machine_hardware_platform(clan_dir, machine_name)
+    return HardwareInfo(system)
+
+
+@API.register
+def show_machine_deployment_target(
+    clan_dir: str | Path, machine_name: str
+) -> str | None:
+    """
+    Show hardware information for a machine returns None if none exist.
+    """
+    config = nix_config()
+    system = config["system"]
+    cmd = nix_eval(
+        [
+            f"{clan_dir}#clanInternals.machines.{system}.{machine_name}",
+            "--apply",
+            "machine: { inherit (machine.config.clan.networking) targetHost; }",
+            "--json",
+        ]
+    )
+    proc = run_no_stdout(cmd)
+    res = proc.stdout.strip()
+
+    target_host = json.loads(res)
+    return target_host.get("targetHost", None)
+
+
+@API.register
+def show_machine_hardware_platform(
+    clan_dir: str | Path, machine_name: str
+) -> str | None:
+    """
+    Show hardware information for a machine returns None if none exist.
+    """
+    config = nix_config()
+    system = config["system"]
+    cmd = nix_eval(
+        [
+            f"{clan_dir}#clanInternals.machines.{system}.{machine_name}",
+            "--apply",
+            "machine: { inherit (machine.config.nixpkgs.hostPlatform) system; }",
+            "--json",
+        ]
+    )
+    proc = run_no_stdout(cmd)
+    res = proc.stdout.strip()
+
+    host_platform = json.loads(res)
+    return host_platform.get("system", None)
+
+
 @API.register
 def generate_machine_hardware_info(
     clan_dir: str | Path,
@@ -63,9 +129,7 @@ def generate_machine_hardware_info(
     hw_file.parent.mkdir(parents=True, exist_ok=True)
 
     # Check if the hardware-configuration.nix file is a template
-    is_template = False
-    if hw_file.exists():
-        is_template = "throw" in hw_file.read_text()
+    is_template = hw_file.exists() and "throw" in hw_file.read_text()
 
     if hw_file.exists() and not force and not is_template:
         raise ClanError(
@@ -78,25 +142,8 @@ def generate_machine_hardware_info(
         f.write(out.stdout)
         print(f"Successfully generated: {hw_file}")
 
-    # TODO: This could be its own API function?
-    config = nix_config()
-    system = config["system"]
-    cmd = nix_eval(
-        [
-            f"{clan_dir}#clanInternals.machines.{system}.{machine_name}",
-            "--apply",
-            "machine: { inherit (machine.config.nixpkgs.hostPlatform) system; }",
-            "--json",
-        ]
-    )
-    proc = run_no_stdout(cmd)
-    res = proc.stdout.strip()
-
-    host_platform = json.loads(res)
-
-    return HardwareInfo(
-        system=host_platform.get("system", None),
-    )
+    system = show_machine_hardware_platform(clan_dir, machine_name)
+    return HardwareInfo(system)
 
 
 def hw_generate_command(args: argparse.Namespace) -> None: