organize files and classes consistently
All checks were successful
checks-impure / test (pull_request) Successful in 1m32s
checks / test (pull_request) Successful in 2m31s

This commit is contained in:
Johannes Kirschbauer 2024-01-20 10:11:52 +01:00
parent ceebfccc82
commit cd2125074f
Signed by: hsjobeki
SSH Key Fingerprint: SHA256:vX3utDqig7Ph5L0JPv87ZTPb/w7cMzREKVZzzLFg9qU
12 changed files with 150 additions and 170 deletions

View File

@ -1,10 +1,10 @@
import argparse
from .app import (
register_join_parser,
register_overview_parser,
show_overview,
)
from clan_cli.clan_uri import ClanURI
from clan_vm_manager.models.interfaces import ClanConfig
from .app import MainApplication
def main() -> None:
@ -24,3 +24,26 @@ def main() -> None:
parser.set_defaults(func=show_overview)
args = parser.parse_args()
args.func(args)
def show_join(args: argparse.Namespace) -> None:
app = MainApplication(
config=ClanConfig(url=args.clan_uri, initial_view="join.trust"),
)
return app.run()
def register_join_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument("clan_uri", type=ClanURI, help="clan URI to join")
parser.set_defaults(func=show_join)
def show_overview(args: argparse.Namespace) -> None:
app = MainApplication(
config=ClanConfig(url=None, initial_view="list"),
)
return app.run()
def register_overview_parser(parser: argparse.ArgumentParser) -> None:
parser.set_defaults(func=show_overview)

View File

@ -1,62 +1,21 @@
#!/usr/bin/env python3
import argparse
from dataclasses import dataclass
from pathlib import Path
import gi
from clan_vm_manager.interfaces import InitialJoinValues
from clan_vm_manager.model.use_views import Views
from clan_vm_manager.views.list import ClanList
from clan_vm_manager.views.trust_join import Trust
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
from clan_cli.clan_uri import ClanURI
from gi.repository import Adw, Gdk, Gio, Gtk
from clan_vm_manager.models.interfaces import ClanConfig
from clan_vm_manager.models.use_vms import VMS
from .constants import constants
from .model.use_vms import VMS
from .windows.main_window import MainWindow
@dataclass
class ClanConfig:
initial_view: str
url: ClanURI | None
class MainWindow(Adw.ApplicationWindow):
def __init__(self, config: ClanConfig) -> None:
super().__init__()
self.set_title("cLAN Manager")
self.set_default_size(980, 650)
view = Adw.ToolbarView()
self.set_content(view)
header = Adw.HeaderBar()
view.add_top_bar(header)
# Initialize all views
stack_view = Views.use().view
stack_view.add_named(ClanList(), "list")
stack_view.add_named(
Trust(initial_values=InitialJoinValues(url=config.url)), "join.trust"
)
stack_view.set_visible_child_name(config.initial_view)
clamp = Adw.Clamp()
clamp.set_child(stack_view)
clamp.set_maximum_size(1000)
view.set_content(clamp)
# Push the first page to the navigation view
class Application(Adw.Application):
class MainApplication(Adw.Application):
def __init__(self, config: ClanConfig) -> None:
super().__init__(
application_id=constants["APPID"], flags=Gio.ApplicationFlags.FLAGS_NONE
@ -84,26 +43,3 @@ class Application(Adw.Application):
css_provider,
Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
)
def show_join(args: argparse.Namespace) -> None:
app = Application(
config=ClanConfig(url=args.clan_uri, initial_view="join.trust"),
)
return app.run()
def register_join_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument("clan_uri", type=ClanURI, help="clan URI to join")
parser.set_defaults(func=show_join)
def show_overview(args: argparse.Namespace) -> None:
app = Application(
config=ClanConfig(url=None, initial_view="list"),
)
return app.run()
def register_overview_parser(parser: argparse.ArgumentParser) -> None:
parser.set_defaults(func=show_overview)

View File

@ -1,26 +0,0 @@
from collections.abc import Callable
from dataclasses import dataclass
from clan_cli.clan_uri import ClanURI
# url is only set, if the app was started with "join <url>"
# Url is usually None, when user clicks "New" clan
@dataclass
class InitialJoinValues:
url: ClanURI | None
@dataclass
class InitialFlashValues:
selected: str | None
@dataclass
class Callbacks:
show_list: Callable[[], None]
show_join: Callable[[], None]
show_flash: Callable[[], None]
spawn_vm: Callable[[str, str], None]
stop_vm: Callable[[str, str], None]
running_vms: Callable[[], list[str]]

View File

@ -1,53 +0,0 @@
from collections.abc import Callable
from typing import Any
from gi.repository import Gio
from clan_vm_manager.models import VM, get_initial_vms
class VMS:
"""
This is a singleton.
It is initialized with the first call of use()
Usage:
VMS.use().get_running_vms()
VMS.use() can also be called before the data is needed. e.g. to eliminate/reduce waiting time.
"""
list_store: Gio.ListStore
_instance: "None | VMS" = None
# Make sure the VMS class is used as a singleton
def __init__(self) -> None:
raise RuntimeError("Call use() instead")
@classmethod
def use(cls: Any) -> "VMS":
if cls._instance is None:
print("Creating new instance")
cls._instance = cls.__new__(cls)
cls.list_store = Gio.ListStore.new(VM)
for vm in get_initial_vms():
cls.list_store.append(vm)
return cls._instance
def handle_vm_stopped(self, func: Callable[[VM, VM], None]) -> None:
for vm in self.list_store:
vm.connect("vm_stopped", func)
def handle_vm_started(self, func: Callable[[VM, VM], None]) -> None:
for vm in self.list_store:
vm.connect("vm_started", func)
def get_running_vms(self) -> list[VM]:
return list(filter(lambda vm: vm.is_running(), self.list_store))
def kill_all(self) -> None:
for vm in self.get_running_vms():
vm.stop()

View File

@ -0,0 +1,23 @@
from dataclasses import dataclass
from enum import StrEnum
import gi
from clan_cli.clan_uri import ClanURI
gi.require_version("Gtk", "4.0")
@dataclass
class InitialJoinValues:
url: ClanURI | None
@dataclass
class ClanConfig:
initial_view: str
url: ClanURI | None
class VMStatus(StrEnum):
RUNNING = "Running"
STOPPED = "Stopped"

View File

@ -1,31 +1,26 @@
import sys
import tempfile
import weakref
from enum import StrEnum
from collections.abc import Callable
from pathlib import Path
from typing import ClassVar
from typing import Any, ClassVar
import gi
from clan_cli import vms
from clan_cli.errors import ClanError
from clan_cli.history.add import HistoryEntry
from clan_cli.history.list import list_history
from gi.repository import GObject
from clan_vm_manager import assets
from clan_vm_manager.errors.show_error import show_error_dialog
from clan_vm_manager.models.interfaces import VMStatus
from .errors.show_error import show_error_dialog
from .executor import MPProcess, spawn
gi.require_version("Gtk", "4.0")
import threading
from gi.repository import GLib
class VMStatus(StrEnum):
RUNNING = "Running"
STOPPED = "Stopped"
from gi.repository import Gio, GLib, GObject
class VM(GObject.Object):
@ -82,7 +77,7 @@ class VM(GObject.Object):
return False
def get_id(self) -> str:
return self.data.flake.flake_url + self.data.flake.flake_attr
return f"{self.data.flake.flake_url}#{self.data.flake.flake_attr}"
def stop_async(self) -> None:
threading.Thread(target=self.stop).start()
@ -101,6 +96,53 @@ class VM(GObject.Object):
return self.process.out_file.read_text()
class VMS:
"""
This is a singleton.
It is initialized with the first call of use()
Usage:
VMS.use().get_running_vms()
VMS.use() can also be called before the data is needed. e.g. to eliminate/reduce waiting time.
"""
list_store: Gio.ListStore
_instance: "None | VMS" = None
# Make sure the VMS class is used as a singleton
def __init__(self) -> None:
raise RuntimeError("Call use() instead")
@classmethod
def use(cls: Any) -> "VMS":
if cls._instance is None:
print("Creating new instance")
cls._instance = cls.__new__(cls)
cls.list_store = Gio.ListStore.new(VM)
for vm in get_initial_vms():
cls.list_store.append(vm)
return cls._instance
def handle_vm_stopped(self, func: Callable[[VM, VM], None]) -> None:
for vm in self.list_store:
vm.connect("vm_stopped", func)
def handle_vm_started(self, func: Callable[[VM, VM], None]) -> None:
for vm in self.list_store:
vm.connect("vm_started", func)
def get_running_vms(self) -> list[VM]:
return list(filter(lambda vm: vm.is_running(), self.list_store))
def kill_all(self) -> None:
for vm in self.get_running_vms():
vm.stop()
def get_initial_vms() -> list[VM]:
vm_list = []
@ -113,7 +155,7 @@ def get_initial_vms() -> list[VM]:
icon = entry.flake.icon
base = VM(
icon=icon,
icon=Path(icon),
status=VMStatus.STOPPED,
data=entry,
)

View File

@ -2,12 +2,10 @@ from functools import partial
import gi
from ..model.use_vms import VMS
gi.require_version("Adw", "1")
from gi.repository import Adw, Gdk, Gtk
from ..models import VM
from clan_vm_manager.models.use_vms import VM, VMS
class ClanList(Gtk.Box):

View File

@ -6,8 +6,7 @@ from clan_cli.errors import ClanError
from clan_cli.history.add import add_history
from clan_vm_manager.errors.show_error import show_error_dialog
from ..interfaces import InitialJoinValues
from clan_vm_manager.models.interfaces import InitialJoinValues
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")

View File

@ -0,0 +1,38 @@
import gi
from clan_vm_manager.models.interfaces import ClanConfig, InitialJoinValues
from clan_vm_manager.models.use_views import Views
from clan_vm_manager.views.list import ClanList
from clan_vm_manager.views.trust_join import Trust
gi.require_version("Adw", "1")
from gi.repository import Adw
class MainWindow(Adw.ApplicationWindow):
def __init__(self, config: ClanConfig) -> None:
super().__init__()
self.set_title("cLAN Manager")
self.set_default_size(980, 650)
view = Adw.ToolbarView()
self.set_content(view)
header = Adw.HeaderBar()
view.add_top_bar(header)
# Initialize all views
stack_view = Views.use().view
stack_view.add_named(ClanList(), "list")
stack_view.add_named(
Trust(initial_values=InitialJoinValues(url=config.url)), "join.trust"
)
stack_view.set_visible_child_name(config.initial_view)
clamp = Adw.Clamp()
clamp.set_child(stack_view)
clamp.set_maximum_size(1000)
view.set_content(clamp)