forked from clan/clan-core
Finished clan history command. Ported remaining async to sync funcs
This commit is contained in:
parent
962a5641b4
commit
5c143bf2da
@ -1,8 +1,8 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import shlex
|
||||
from collections.abc import Callable, Coroutine
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from subprocess import PIPE, Popen
|
||||
from typing import Any, NamedTuple
|
||||
|
||||
from .custom_logger import get_caller
|
||||
@ -17,7 +17,7 @@ class CmdOut(NamedTuple):
|
||||
cwd: Path | None = None
|
||||
|
||||
|
||||
async def run(cmd: list[str], cwd: Path | None = None) -> CmdOut:
|
||||
def run(cmd: list[str], cwd: Path | None = None) -> CmdOut:
|
||||
cwd_res = None
|
||||
if cwd is not None:
|
||||
if not cwd.exists():
|
||||
@ -28,13 +28,14 @@ async def run(cmd: list[str], cwd: Path | None = None) -> CmdOut:
|
||||
log.debug(
|
||||
f"Command: {shlex.join(cmd)}\nWorking directory: {cwd_res}\nCaller : {get_caller()}"
|
||||
)
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
proc = Popen(
|
||||
args=cmd,
|
||||
stderr=PIPE,
|
||||
stdout=PIPE,
|
||||
text=True,
|
||||
cwd=cwd_res,
|
||||
)
|
||||
stdout, stderr = await proc.communicate()
|
||||
stdout, stderr = proc.communicate()
|
||||
|
||||
if proc.returncode != 0:
|
||||
raise ClanError(
|
||||
@ -43,20 +44,18 @@ command: {shlex.join(cmd)}
|
||||
working directory: {cwd_res}
|
||||
exit code: {proc.returncode}
|
||||
stderr:
|
||||
{stderr.decode("utf-8")}
|
||||
{stderr}
|
||||
stdout:
|
||||
{stdout.decode("utf-8")}
|
||||
{stdout}
|
||||
"""
|
||||
)
|
||||
|
||||
return CmdOut(stdout.decode("utf-8"), stderr.decode("utf-8"), cwd=cwd)
|
||||
return CmdOut(stdout, stderr, cwd=cwd)
|
||||
|
||||
|
||||
def runforcli(
|
||||
func: Callable[..., Coroutine[Any, Any, dict[str, CmdOut]]], *args: Any
|
||||
) -> None:
|
||||
def runforcli(func: Callable[..., dict[str, CmdOut]], *args: Any) -> None:
|
||||
try:
|
||||
res = asyncio.run(func(*args))
|
||||
res = func(*args)
|
||||
|
||||
for name, out in res.items():
|
||||
if out.stderr:
|
@ -2,14 +2,14 @@
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
from ..async_cmd import CmdOut, run, runforcli
|
||||
from ..cmd import CmdOut, run, runforcli
|
||||
from ..errors import ClanError
|
||||
from ..nix import nix_command, nix_shell
|
||||
|
||||
DEFAULT_URL: str = "git+https://git.clan.lol/clan/clan-core?new-clan"
|
||||
|
||||
|
||||
async def create_flake(directory: Path, url: str) -> dict[str, CmdOut]:
|
||||
def create_flake(directory: Path, url: str) -> dict[str, CmdOut]:
|
||||
if not directory.exists():
|
||||
directory.mkdir()
|
||||
else:
|
||||
@ -23,36 +23,27 @@ async def create_flake(directory: Path, url: str) -> dict[str, CmdOut]:
|
||||
url,
|
||||
]
|
||||
)
|
||||
out = await run(command, cwd=directory)
|
||||
out = run(command, cwd=directory)
|
||||
response["flake init"] = out
|
||||
|
||||
command = nix_shell(["nixpkgs#git"], ["git", "init"])
|
||||
out = await run(command, cwd=directory)
|
||||
out = run(command, cwd=directory)
|
||||
response["git init"] = out
|
||||
|
||||
command = nix_shell(["nixpkgs#git"], ["git", "add", "."])
|
||||
out = await run(command, cwd=directory)
|
||||
out = run(command, cwd=directory)
|
||||
response["git add"] = out
|
||||
|
||||
# command = nix_shell(["nixpkgs#git"], ["git", "config", "init.defaultBranch", "main"])
|
||||
# out = await run(command, cwd=directory)
|
||||
# response["git config"] = out
|
||||
|
||||
command = nix_shell(["nixpkgs#git"], ["git", "config", "user.name", "clan-tool"])
|
||||
out = await run(command, cwd=directory)
|
||||
out = run(command, cwd=directory)
|
||||
response["git config"] = out
|
||||
|
||||
command = nix_shell(
|
||||
["nixpkgs#git"], ["git", "config", "user.email", "clan@example.com"]
|
||||
)
|
||||
out = await run(command, cwd=directory)
|
||||
out = run(command, cwd=directory)
|
||||
response["git config"] = out
|
||||
|
||||
# TODO: Find out why this fails on Johannes machine
|
||||
# command = nix_shell(["nixpkgs#git"], ["git", "commit", "-a", "-m", "Initial commit"])
|
||||
# out = await run(command, cwd=directory)
|
||||
# response["git commit"] = out
|
||||
|
||||
return response
|
||||
|
||||
|
||||
|
@ -6,7 +6,7 @@ from pathlib import Path
|
||||
|
||||
from ..errors import ClanError
|
||||
from ..machines.list import list_machines
|
||||
from ..nix import nix_config, nix_eval, nix_metadata
|
||||
from ..nix import nix_build, nix_config, nix_eval, nix_metadata
|
||||
|
||||
|
||||
@dataclass
|
||||
@ -14,6 +14,7 @@ class FlakeConfig:
|
||||
flake_url: str | Path
|
||||
flake_attr: str
|
||||
|
||||
clan_name: str
|
||||
nar_hash: str
|
||||
icon: str | None
|
||||
description: str | None
|
||||
@ -56,10 +57,52 @@ stderr:
|
||||
else:
|
||||
icon_path = res.strip('"')
|
||||
|
||||
if not Path(icon_path).exists():
|
||||
cmd = nix_build(
|
||||
[
|
||||
f'{flake_url}#clanInternals.machines."{system}"."{flake_attr}".config.clanCore.clanIcon'
|
||||
]
|
||||
)
|
||||
proc = subprocess.run(cmd, text=True, capture_output=True)
|
||||
assert proc.stdout is not None
|
||||
if proc.returncode != 0:
|
||||
raise ClanError(
|
||||
f"""
|
||||
command: {shlex.join(cmd)}
|
||||
exit code: {proc.returncode}
|
||||
stdout:
|
||||
{proc.stdout}
|
||||
stderr:
|
||||
{proc.stderr}
|
||||
"""
|
||||
)
|
||||
|
||||
cmd = nix_eval(
|
||||
[
|
||||
f'{flake_url}#clanInternals.machines."{system}"."{flake_attr}".config.clanCore.clanName'
|
||||
]
|
||||
)
|
||||
|
||||
proc = subprocess.run(cmd, text=True, capture_output=True)
|
||||
assert proc.stdout is not None
|
||||
if proc.returncode != 0:
|
||||
raise ClanError(
|
||||
f"""
|
||||
command: {shlex.join(cmd)}
|
||||
exit code: {proc.returncode}
|
||||
stdout:
|
||||
{proc.stdout}
|
||||
stderr:
|
||||
{proc.stderr}
|
||||
"""
|
||||
)
|
||||
clan_name = proc.stdout.strip().strip('"')
|
||||
|
||||
meta = nix_metadata(flake_url)
|
||||
|
||||
return FlakeConfig(
|
||||
flake_url=flake_url,
|
||||
clan_name=clan_name,
|
||||
flake_attr=flake_attr,
|
||||
nar_hash=meta["locked"]["narHash"],
|
||||
icon=icon_path,
|
||||
@ -83,6 +126,7 @@ def inspect_command(args: argparse.Namespace) -> None:
|
||||
res = inspect_flake(
|
||||
flake_url=inspect_options.flake, flake_attr=inspect_options.machine
|
||||
)
|
||||
print("cLAN name:", res.clan_name)
|
||||
print("Icon:", res.icon)
|
||||
print("Description:", res.description)
|
||||
print("Last updated:", res.last_updated)
|
||||
|
@ -39,7 +39,7 @@ def list_history() -> list[HistoryEntry]:
|
||||
content: str = f.read()
|
||||
parsed: list[dict] = json.loads(content)
|
||||
logs = [HistoryEntry(**p) for p in parsed]
|
||||
except json.JSONDecodeError as ex:
|
||||
except (json.JSONDecodeError, TypeError) as ex:
|
||||
print("Failed to load history. Invalid JSON.")
|
||||
print(f"{user_history_file()}: {ex}")
|
||||
|
||||
|
@ -4,7 +4,7 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import gi
|
||||
from clan_cli import flakes, history, vms
|
||||
from clan_cli import history, vms
|
||||
|
||||
gi.require_version("GdkPixbuf", "2.0")
|
||||
from gi.repository import GdkPixbuf
|
||||
@ -18,7 +18,6 @@ class VMBase:
|
||||
name: str
|
||||
url: str
|
||||
status: bool
|
||||
_path: Path
|
||||
|
||||
@staticmethod
|
||||
def name_to_type_map() -> OrderedDict[str, type]:
|
||||
@ -28,7 +27,6 @@ class VMBase:
|
||||
"Name": str,
|
||||
"URL": str,
|
||||
"Online": bool,
|
||||
"_Path": str,
|
||||
}
|
||||
)
|
||||
|
||||
@ -43,7 +41,6 @@ class VMBase:
|
||||
"Name": self.name,
|
||||
"URL": self.url,
|
||||
"Online": self.status,
|
||||
"_Path": str(self._path),
|
||||
}
|
||||
)
|
||||
|
||||
@ -52,16 +49,10 @@ class VMBase:
|
||||
import asyncio
|
||||
|
||||
# raise Exception("Cannot run VMs yet")
|
||||
vm = asyncio.run(
|
||||
vms.run.inspect_vm(flake_url=self._path, flake_attr="defaultVM")
|
||||
)
|
||||
vm = asyncio.run(vms.run.inspect_vm(flake_url=self.url, flake_attr="defaultVM"))
|
||||
vms.run.run_vm(vm)
|
||||
|
||||
|
||||
# for line in task.log_lines():
|
||||
# print(line, end="")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VM:
|
||||
# Inheritance is bad. Lets use composition
|
||||
@ -75,32 +66,19 @@ class VM:
|
||||
def get_initial_vms(start: int = 0, end: int | None = None) -> list[VM]:
|
||||
vm_list = []
|
||||
|
||||
# TODO: list_history() should return a list of dicts, not a list of paths
|
||||
# Execute `clan flakes add <path>` to democlan for this to work
|
||||
for entry in history.list.list_history():
|
||||
flake_config = flakes.inspect.inspect_flake(entry.path, "defaultVM")
|
||||
vm_config = vms.inspect.inspect_vm(entry.path, "defaultVM")
|
||||
|
||||
# if flake_config.icon is None:
|
||||
# icon = assets.loc / "placeholder.jpeg"
|
||||
# else:
|
||||
# icon = flake_config.icon
|
||||
icon = assets.loc / "placeholder.jpeg"
|
||||
# TODO: clan flakes inspect currently points to an icon that doesn't exist
|
||||
# the reason being that the icon is not in the nix store, as the democlan has
|
||||
# not been built yet. Not sure how to handle this.
|
||||
# I think how to do this is to add democlan as a flake.nix dependency and then
|
||||
# put it into the devshell.
|
||||
if entry.flake.icon is not None:
|
||||
icon = entry.flake.icon
|
||||
|
||||
print(f"Icon: {icon}")
|
||||
new_vm = {
|
||||
"icon": icon,
|
||||
"name": vm_config.clan_name,
|
||||
"url": flake_config.flake_url,
|
||||
"_path": entry.path,
|
||||
"status": False,
|
||||
}
|
||||
vm_list.append(VM(base=VMBase(**new_vm)))
|
||||
base = VMBase(
|
||||
icon=icon,
|
||||
name=entry.flake.clan_name,
|
||||
url=entry.flake.flake_url,
|
||||
status=False,
|
||||
)
|
||||
vm_list.append(VM(base=base))
|
||||
|
||||
# start/end slices can be used for pagination
|
||||
return vm_list[start:end]
|
||||
|
Loading…
Reference in New Issue
Block a user