Merge pull request 'history: added vm config to FlakeConfig' (#671) from Qubasa-origin/Qubasa-main-no-flake-update into main

pkgs/clan-cli/clan_cli/flakes/inspect.py

@@ -8,6 +8,7 @@ from ..dirs import specific_groot_dir
 from ..errors import ClanError
 from ..machines.list import list_machines
 from ..nix import nix_build, nix_config, nix_eval, nix_metadata
+from ..vms.inspect import VmConfig, inspect_vm
 
 
 @dataclass
@@ -21,6 +22,7 @@ class FlakeConfig:
     description: str | None
     last_updated: str
     revision: str | None
+    vm: VmConfig
 
 
 def run_cmd(cmd: list[str]) -> str:
@@ -50,6 +52,8 @@ def inspect_flake(flake_url: str | Path, flake_attr: str) -> FlakeConfig:
             f"Machine {flake_attr} not found in {flake_url}. Available machines: {', '.join(machines)}"
         )
 
+    vm = inspect_vm(flake_url, flake_attr)
+
     # Get the cLAN name
     cmd = nix_eval(
         [
@@ -86,6 +90,7 @@ def inspect_flake(flake_url: str | Path, flake_attr: str) -> FlakeConfig:
     meta = nix_metadata(flake_url)
 
     return FlakeConfig(
+        vm=vm,
         flake_url=flake_url,
         clan_name=clan_name,
         flake_attr=flake_attr,

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

@@ -9,7 +9,7 @@ from clan_cli.flakes.inspect import FlakeConfig, inspect_flake
 
 from ..clan_uri import ClanURI
 from ..dirs import user_history_file
-from ..locked_open import locked_open
+from ..locked_open import read_history_file, write_history_file
 
 
 class EnhancedJSONEncoder(json.JSONEncoder):
@@ -23,29 +23,46 @@ class EnhancedJSONEncoder(json.JSONEncoder):
 class HistoryEntry:
     last_used: str
     flake: FlakeConfig
+    settings: dict[str, Any] = dataclasses.field(default_factory=dict)
 
     def __post_init__(self) -> None:
         if isinstance(self.flake, dict):
             self.flake = FlakeConfig(**self.flake)
 
 
+def merge_dicts(d1: dict, d2: dict) -> dict:
+    # create a new dictionary that copies d1
+    merged = dict(d1)
+    # iterate over the keys and values of d2
+    for key, value in d2.items():
+        # if the key is in d1 and both values are dictionaries, merge them recursively
+        if key in d1 and isinstance(d1[key], dict) and isinstance(value, dict):
+            merged[key] = merge_dicts(d1[key], value)
+        # otherwise, update the value of the key in the merged dictionary
+        else:
+            merged[key] = value
+    # return the merged dictionary
+    return merged
+
+
 def list_history() -> list[HistoryEntry]:
     logs: list[HistoryEntry] = []
     if not user_history_file().exists():
         return []
 
-    with locked_open(user_history_file(), "r") as f:
-        try:
-            content: str = f.read()
-            parsed: list[dict] = json.loads(content)
-            logs = [HistoryEntry(**p) for p in parsed]
-        except (json.JSONDecodeError, TypeError) as ex:
-            print("Failed to load history. Invalid JSON.")
-            print(f"{user_history_file()}: {ex}")
+    try:
+        parsed = read_history_file()
+        for i, p in enumerate(parsed.copy()):
+            parsed[i] = merge_dicts(p, p["settings"])
+        logs = [HistoryEntry(**p) for p in parsed]
+    except (json.JSONDecodeError, TypeError) as ex:
+        print("Failed to load history. Invalid JSON.")
+        print(f"{user_history_file()}: {ex}")
 
     return logs
 
 
+# TODO: Add all vm entries to history
 def add_history(uri: ClanURI) -> list[HistoryEntry]:
     uri.check_exits()
     user_history_file().parent.mkdir(parents=True, exist_ok=True)
@@ -68,9 +85,7 @@ def add_history(uri: ClanURI) -> list[HistoryEntry]:
         )
         logs.append(history)
 
-    with locked_open(user_history_file(), "w+") as f:
-        f.write(json.dumps(logs, cls=EnhancedJSONEncoder, indent=4))
-        f.truncate()
+    write_history_file(logs)
 
     return logs
 

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

@@ -2,12 +2,10 @@
 import argparse
 import copy
 import datetime
-import json
 
-from ..dirs import user_history_file
-from ..locked_open import locked_open
+from ..locked_open import write_history_file
 from ..nix import nix_metadata
-from .add import EnhancedJSONEncoder, HistoryEntry, list_history
+from .add import HistoryEntry, list_history
 
 
 def update_history() -> list[HistoryEntry]:
@@ -29,9 +27,7 @@ def update_history() -> list[HistoryEntry]:
         # TODO: Delete stale entries
         new_logs.append(new_entry)
 
-    with locked_open(user_history_file(), "w+") as f:
-        f.write(json.dumps(new_logs, cls=EnhancedJSONEncoder, indent=4))
-        f.truncate()
+    write_history_file(new_logs)
     return new_logs
 
 

pkgs/clan-cli/clan_cli/locked_open.py

@@ -1,11 +1,23 @@
+import dataclasses
 import fcntl
+import json
 from collections.abc import Generator
 from contextlib import contextmanager
 from pathlib import Path
+from typing import Any
+
+from .dirs import user_history_file
+
+
+class EnhancedJSONEncoder(json.JSONEncoder):
+    def default(self, o: Any) -> Any:
+        if dataclasses.is_dataclass(o):
+            return dataclasses.asdict(o)
+        return super().default(o)
 
 
 @contextmanager
-def locked_open(filename: str | Path, mode: str = "r") -> Generator:
+def _locked_open(filename: str | Path, mode: str = "r") -> Generator:
     """
     This is a context manager that provides an advisory write lock on the file specified by `filename` when entering the context, and releases the lock when leaving the context. The lock is acquired using the `fcntl` module's `LOCK_EX` flag, which applies an exclusive write lock to the file.
     """
@@ -13,3 +25,16 @@ def locked_open(filename: str | Path, mode: str = "r") -> Generator:
         fcntl.flock(fd, fcntl.LOCK_EX)
         yield fd
         fcntl.flock(fd, fcntl.LOCK_UN)
+
+
+def write_history_file(data: Any) -> None:
+    with _locked_open(user_history_file(), "w+") as f:
+        f.write(json.dumps(data, cls=EnhancedJSONEncoder, indent=4))
+        f.truncate()
+
+
+def read_history_file() -> list[dict]:
+    with _locked_open(user_history_file(), "r") as f:
+        content: str = f.read()
+        parsed: list[dict] = json.loads(content)
+        return parsed

pkgs/clan-vm-manager/clan_vm_manager/windows/join.py

@@ -75,6 +75,7 @@ class Trust(Gtk.Box):
         self.set_center_widget(layout)
 
     def on_trust(self, widget: Gtk.Widget) -> None:
+        # TODO: @Johannes, replace image with real clan logo
         try:
             uri = self.url or ClanURI(self.entry.get_text())
             print(f"trusted: {uri}")