clan_vm_manager: Renamed VMs singleton to ClanStore. And VM to VMObject
All checks were successful
checks / check-links (pull_request) Successful in 21s
checks / checks-impure (pull_request) Successful in 1m56s
checks / checks (pull_request) Successful in 2m22s

This commit is contained in:
Luis Hebendanz 2024-03-03 16:47:38 +07:00
parent 359ad22c90
commit 191562a84e
4 changed files with 22 additions and 22 deletions

View File

@ -7,7 +7,7 @@ import gi
from clan_cli.clan_uri import ClanURI
from clan_cli.history.add import HistoryEntry, add_history
from clan_vm_manager.singletons.use_vms import VMs
from clan_vm_manager.singletons.use_vms import ClanStore
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
@ -88,7 +88,7 @@ class JoinList:
def _on_join_finished(self, _source: GObject.Object, value: JoinValue) -> None:
log.info(f"Join finished: {value.url}")
self.discard(value)
VMs.use().push_history_entry(value.entry)
ClanStore.use().push_history_entry(value.entry)
def discard(self, value: JoinValue) -> None:
(has, idx) = self.list_store.find(value)

View File

@ -29,8 +29,8 @@ from gi.repository import GLib, GObject, Gtk
log = logging.getLogger(__name__)
class VM(GObject.Object):
__gtype_name__: ClassVar = "VMGobject"
class VMObject(GObject.Object):
__gtype_name__: ClassVar = "VMObject"
# Define a custom signal with the name "vm_stopped" and a string argument for the message
__gsignals__: ClassVar = {
"vm_status_changed": (GObject.SignalFlags.RUN_FIRST, None, [GObject.Object])
@ -85,7 +85,7 @@ class VM(GObject.Object):
# Make sure the VM is killed when the reference to this object is dropped
self._finalizer = weakref.finalize(self, self.kill_ref_drop)
def on_vm_status_changed(self, vm: "VM", _vm: "VM") -> None:
def on_vm_status_changed(self, vm: "VMObject", _vm: "VMObject") -> None:
self.switch.set_state(self.is_running() and not self.is_building())
if self.switch.get_sensitive() is False and not self.is_building():
self.switch.set_sensitive(True)
@ -317,11 +317,11 @@ class VMStore(GKVStore):
__gtype_name__ = "MyVMStore"
def __init__(self) -> None:
super().__init__(VM, lambda vm: vm.data.flake.flake_attr)
super().__init__(VMObject, lambda vm: vm.data.flake.flake_attr)
class VMs:
_instance: "None | VMs" = None
class ClanStore:
_instance: "None | ClanStore" = None
_clan_store: GKVStore[str, VMStore]
# Make sure the VMS class is used as a singleton
@ -329,7 +329,7 @@ class VMs:
raise RuntimeError("Call use() instead")
@classmethod
def use(cls: Any) -> "VMs":
def use(cls: Any) -> "ClanStore":
if cls._instance is None:
cls._instance = cls.__new__(cls)
cls._clan_store = GKVStore(
@ -353,13 +353,13 @@ class VMs:
else:
icon = entry.flake.icon
vm = VM(
vm = VMObject(
icon=Path(icon),
data=entry,
)
self.push(vm)
def push(self, vm: VM) -> None:
def push(self, vm: VMObject) -> None:
url = vm.data.flake.flake_url
# Only write to the store if the VM is not already in it
@ -374,16 +374,16 @@ class VMs:
vm_store = self.clan_store[url]
vm_store.append(vm)
def remove(self, vm: VM) -> None:
def remove(self, vm: VMObject) -> None:
del self.clan_store[vm.data.flake.flake_url][vm.data.flake.flake_attr]
def get_vm(self, flake_url: str, flake_attr: str) -> None | VM:
def get_vm(self, flake_url: str, flake_attr: str) -> None | VMObject:
clan = self.clan_store.get(flake_url)
if clan is None:
return None
return clan.get(flake_attr, None)
def get_running_vms(self) -> list[VM]:
def get_running_vms(self) -> list[VMObject]:
return [
vm
for clan in self.clan_store.values()

View File

@ -9,7 +9,7 @@ from clan_cli.clan_uri import ClanURI
from clan_vm_manager.components.interfaces import ClanConfig
from clan_vm_manager.singletons.use_join import JoinList, JoinValue
from clan_vm_manager.singletons.use_vms import VM, VMs, VMStore
from clan_vm_manager.singletons.use_vms import ClanStore, VMObject, VMStore
gi.require_version("Adw", "1")
from gi.repository import Adw, Gdk, Gio, GLib, GObject, Gtk
@ -60,7 +60,7 @@ class ClanList(Gtk.Box):
self.append(self.join_boxed_list)
self.group_list = create_boxed_list(
model=VMs.use().clan_store, render_row=self.render_group_row
model=ClanStore.use().clan_store, render_row=self.render_group_row
)
self.group_list.add_css_class("group-list")
self.append(self.group_list)
@ -104,7 +104,7 @@ class ClanList(Gtk.Box):
target = parameter.get_string()
print("Adding new machine", target)
def render_vm_row(self, boxed_list: Gtk.ListBox, vm: VM) -> Gtk.Widget:
def render_vm_row(self, boxed_list: Gtk.ListBox, vm: VMObject) -> Gtk.Widget:
# Remove no-shadow class if attached
if boxed_list.has_css_class("no-shadow"):
boxed_list.remove_css_class("no-shadow")
@ -189,7 +189,7 @@ class ClanList(Gtk.Box):
row.set_subtitle(item.url.get_internal())
row.add_css_class("trust")
if item.url.params.flake_attr in VMs.use().clan_store:
if item.url.params.flake_attr in ClanStore.use().clan_store:
sub = row.get_subtitle()
row.set_subtitle(
sub + "\nClan already exists. Joining again will update it"

View File

@ -7,7 +7,7 @@ from clan_cli.history.list import list_history
from clan_vm_manager.components.interfaces import ClanConfig
from clan_vm_manager.singletons.use_views import Views
from clan_vm_manager.singletons.use_vms import VMs
from clan_vm_manager.singletons.use_vms import ClanStore
from clan_vm_manager.views.details import Details
from clan_vm_manager.views.list import ClanList
@ -35,7 +35,7 @@ class MainWindow(Adw.ApplicationWindow):
app = Gio.Application.get_default()
self.tray_icon: TrayIcon = TrayIcon(app)
# Initialize all VMs
# Initialize all ClanStore
threading.Thread(target=self._populate_vms).start()
# Initialize all views
@ -63,8 +63,8 @@ class MainWindow(Adw.ApplicationWindow):
# Execute `clan flakes add <path>` to democlan for this to work
# TODO: Make list_history a generator function
for entry in list_history():
GLib.idle_add(VMs.use().create_vm_task, entry)
GLib.idle_add(ClanStore.use().create_vm_task, entry)
def on_destroy(self, *_args: Any) -> None:
self.tray_icon.destroy()
VMs.use().kill_all()
ClanStore.use().kill_all()