clan ui: move print into log.debug statements

This commit is contained in:
Johannes Kirschbauer 2024-05-20 19:59:50 +02:00
parent 8687801cee
commit d174fbd445
Signed by: hsjobeki
SSH Key Fingerprint: SHA256:vX3utDqig7Ph5L0JPv87ZTPb/w7cMzREKVZzzLFg9qU
5 changed files with 38 additions and 49 deletions

View File

@ -1,13 +1,8 @@
from clan_cli import create_parser
from clan_cli.api import API
from clan_cli.api.schema_compat import to_json_schema
def main() -> None:
# Create the parser to register the API functions
create_parser()
schema = to_json_schema(API._registry)
schema = API.to_json_schema()
print(
f"""export const schema = {schema} as const;
"""

View File

@ -9,5 +9,35 @@ class _MethodRegistry:
self._registry[fn.__name__] = fn
return fn
def to_json_schema(self) -> str:
# Import only when needed
import json
from typing import get_type_hints
from clan_cli.api.util import type_to_dict
api_schema = {
"$comment": "An object containing API methods. ",
"type": "object",
"additionalProperties": False,
"required": ["list_machines"],
"properties": {},
}
for name, func in self._registry.items():
hints = get_type_hints(func)
serialized_hints = {
"argument" if key != "return" else "return": type_to_dict(
value, scope=name + " argument" if key != "return" else "return"
)
for key, value in hints.items()
}
api_schema["properties"][name] = {
"type": "object",
"required": [k for k in serialized_hints.keys()],
"additionalProperties": False,
"properties": {**serialized_hints},
}
return json.dumps(api_schema, indent=2)
API = _MethodRegistry()

View File

@ -1,16 +1,10 @@
import dataclasses
import json
from types import NoneType, UnionType
from typing import Any, Callable, Union, get_type_hints
import pathlib
def type_to_dict(t: Any, scope: str = "") -> dict:
# print(
# f"Type: {t}, Scope: {scope}, has origin: {hasattr(t, '__origin__')} ",
# type(t) is UnionType,
# )
if t is None:
return {"type": "null"}
@ -83,29 +77,3 @@ def type_to_dict(t: Any, scope: str = "") -> dict:
raise BaseException(f"Error primitive type not supported {str(t)}")
else:
raise BaseException(f"Error type not supported {str(t)}")
def to_json_schema(methods: dict[str, Callable]) -> str:
api_schema = {
"$comment": "An object containing API methods. ",
"type": "object",
"additionalProperties": False,
"required": ["list_machines"],
"properties": {},
}
for name, func in methods.items():
hints = get_type_hints(func)
serialized_hints = {
"argument" if key != "return" else "return": type_to_dict(
value, scope=name + " argument" if key != "return" else "return"
)
for key, value in hints.items()
}
api_schema["properties"][name] = {
"type": "object",
"required": [k for k in serialized_hints.keys()],
"additionalProperties": False,
"properties": {**serialized_hints},
}
return json.dumps(api_schema, indent=2)

View File

@ -12,7 +12,6 @@ log = logging.getLogger(__name__)
@API.register
def list_machines(flake_url: Path | str) -> list[str]:
print("list_machines", flake_url)
config = nix_config()
system = config["system"]
cmd = nix_eval(

View File

@ -1,5 +1,5 @@
import dataclasses
import json
import logging
import sys
import threading
from threading import Lock
@ -19,6 +19,8 @@ site_index: Path = (
/ Path("clan_vm_manager/.webui/index.html")
).resolve()
log = logging.getLogger(__name__)
class WebView:
def __init__(self, methods: dict[str, Callable]) -> None:
@ -44,8 +46,8 @@ class WebView:
method_name = payload["method"]
handler_fn = self.method_registry[method_name]
print(f"Received message: {payload}")
print(f"Queue size: {self.queue_size} (Wait)")
log.debug(f"Received message: {payload}")
log.debug(f"Queue size: {self.queue_size} (Wait)")
def threaded_wrapper() -> bool:
"""
@ -78,19 +80,14 @@ class WebView:
self, handler_fn: Callable[[Any], Any], data: Any, method_name: str
) -> None:
with self.mutex_lock:
print("Executing", method_name)
print("threading locked ...")
log.debug("Executing... ", method_name)
result = handler_fn(data)
serialized = json.dumps(result)
# Use idle_add to queue the response call to js on the main GTK thread
GLib.idle_add(self.return_data_to_js, method_name, serialized)
print("threading unlocked")
self.queue_size -= 1
if self.queue_size > 0:
print(f"remaining queue size: {self.queue_size}")
else:
print(f"Queue empty")
log.debug(f"Done: Remaining queue size: {self.queue_size}")
def return_data_to_js(self, method_name: str, serialized: str) -> bool:
# This function must be run on the main GTK thread to interact with the webview