clan-vm-manager: Add library for mypy pygobject types #921
@ -33,10 +33,8 @@ class MainApplication(Adw.Application):
|
||||
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
super().__init__(
|
||||
*args,
|
||||
application_id="org.clan.vm-manager",
|
||||
flags=Gio.ApplicationFlags.HANDLES_COMMAND_LINE,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
self.add_main_option(
|
||||
@ -48,7 +46,7 @@ class MainApplication(Adw.Application):
|
||||
None,
|
||||
)
|
||||
|
||||
self.window: Adw.ApplicationWindow | None = None
|
||||
self.window: "MainWindow" | None = None
|
||||
self.connect("activate", self.on_activate)
|
||||
self.connect("shutdown", self.on_shutdown)
|
||||
|
||||
@ -113,8 +111,10 @@ class MainApplication(Adw.Application):
|
||||
log.debug(f"Style css path: {resource_path}")
|
||||
css_provider = Gtk.CssProvider()
|
||||
css_provider.load_from_path(str(resource_path))
|
||||
display = Gdk.Display.get_default()
|
||||
assert display is not None
|
||||
Gtk.StyleContext.add_provider_for_display(
|
||||
Gdk.Display.get_default(),
|
||||
display,
|
||||
css_provider,
|
||||
Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
|
||||
)
|
||||
|
@ -134,8 +134,8 @@ class GKVStore(GObject.GObject, Gio.ListModel, Generic[K, V]):
|
||||
def do_get_item(self, position: int) -> V | None:
|
||||
return self.get_item(position)
|
||||
|
||||
def get_item_type(self) -> GObject.GType:
|
||||
return self.gtype.__gtype__
|
||||
def get_item_type(self) -> Any:
|
||||
return self.gtype.__gtype__ # type: ignore[attr-defined]
|
||||
|
||||
def do_get_item_type(self) -> GObject.GType:
|
||||
return self.get_item_type()
|
||||
@ -187,10 +187,10 @@ class GKVStore(GObject.GObject, Gio.ListModel, Generic[K, V]):
|
||||
return len(self._items)
|
||||
|
||||
# O(1) operation
|
||||
def __getitem__(self, key: K) -> V:
|
||||
def __getitem__(self, key: K) -> V: # type: ignore[override]
|
||||
return self._items[key]
|
||||
|
||||
def __contains__(self, key: K) -> bool:
|
||||
def __contains__(self, key: K) -> bool: # type: ignore[override]
|
||||
return key in self._items
|
||||
|
||||
def __str__(self) -> str:
|
||||
|
@ -24,6 +24,9 @@ import sys
|
||||
from collections.abc import Callable
|
||||
from typing import Any, ClassVar
|
||||
|
||||
import gi
|
||||
|
||||
gi.require_version("Gtk", "4.0")
|
||||
from gi.repository import GdkPixbuf, Gio, GLib, Gtk
|
||||
|
||||
|
||||
@ -168,7 +171,7 @@ class ImplUnavailableError(Exception):
|
||||
|
||||
|
||||
class BaseImplementation:
|
||||
def __init__(self, application: Gtk.Application) -> None:
|
||||
def __init__(self, application: Any) -> None:
|
||||
self.application = application
|
||||
self.menu_items: dict[int, Any] = {}
|
||||
self.menu_item_id: int = 1
|
||||
@ -1090,8 +1093,8 @@ class Win32Implementation(BaseImplementation):
|
||||
|
||||
|
||||
class TrayIcon:
|
||||
def __init__(self, application: Gtk.Application) -> None:
|
||||
self.application: Gtk.Application = application
|
||||
def __init__(self, application: Gio.Application) -> None:
|
||||
self.application: Gio.Application = application
|
||||
self.available: bool = True
|
||||
self.implementation: Any = None
|
||||
|
||||
|
@ -32,13 +32,6 @@ class VMObject(GObject.Object):
|
||||
"vm_status_changed": (GObject.SignalFlags.RUN_FIRST, None, [])
|
||||
}
|
||||
|
||||
def _vm_status_changed_task(self) -> bool:
|
||||
self.emit("vm_status_changed")
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
def update(self, data: HistoryEntry) -> None:
|
||||
self.data = data
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
icon: Path,
|
||||
@ -47,16 +40,20 @@ class VMObject(GObject.Object):
|
||||
super().__init__()
|
||||
|
||||
# Store the data from the history entry
|
||||
self.data = data
|
||||
self.data: HistoryEntry = data
|
||||
|
||||
# Create a process object to store the VM process
|
||||
self.vm_process = MPProcess("vm_dummy", mp.Process(), Path("./dummy"))
|
||||
self.build_process = MPProcess("build_dummy", mp.Process(), Path("./dummy"))
|
||||
self.vm_process: MPProcess = MPProcess(
|
||||
"vm_dummy", mp.Process(), Path("./dummy")
|
||||
)
|
||||
self.build_process: MPProcess = MPProcess(
|
||||
"build_dummy", mp.Process(), Path("./dummy")
|
||||
)
|
||||
self._start_thread: threading.Thread = threading.Thread()
|
||||
self.machine: Machine | None = None
|
||||
|
||||
# Watcher to stop the VM
|
||||
self.KILL_TIMEOUT = 20 # seconds
|
||||
self.KILL_TIMEOUT: int = 20 # seconds
|
||||
self._stop_thread: threading.Thread = threading.Thread()
|
||||
|
||||
# Build progress bar vars
|
||||
@ -66,7 +63,7 @@ class VMObject(GObject.Object):
|
||||
self.prog_bar_id: int = 0
|
||||
|
||||
# Create a temporary directory to store the logs
|
||||
self.log_dir = tempfile.TemporaryDirectory(
|
||||
self.log_dir: tempfile.TemporaryDirectory = tempfile.TemporaryDirectory(
|
||||
prefix="clan_vm-", suffix=f"-{self.data.flake.flake_attr}"
|
||||
)
|
||||
self._logs_id: int = 0
|
||||
@ -75,14 +72,21 @@ class VMObject(GObject.Object):
|
||||
# To be able to set the switch state programmatically
|
||||
# we need to store the handler id returned by the connect method
|
||||
# and block the signal while we change the state. This is cursed.
|
||||
self.switch = Gtk.Switch()
|
||||
self.switch: Gtk.Switch = Gtk.Switch()
|
||||
self.switch_handler_id: int = self.switch.connect(
|
||||
"notify::active", self._on_switch_toggle
|
||||
)
|
||||
self.connect("vm_status_changed", self._on_vm_status_changed)
|
||||
|
||||
# Make sure the VM is killed when the reference to this object is dropped
|
||||
self._finalizer = weakref.finalize(self, self._kill_ref_drop)
|
||||
self._finalizer: weakref.finalize = weakref.finalize(self, self._kill_ref_drop)
|
||||
|
||||
def _vm_status_changed_task(self) -> bool:
|
||||
self.emit("vm_status_changed")
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
def update(self, data: HistoryEntry) -> None:
|
||||
self.data = data
|
||||
|
||||
def _on_vm_status_changed(self, source: "VMObject") -> None:
|
||||
self.switch.set_state(self.is_running() and not self.is_building())
|
||||
@ -93,9 +97,8 @@ class VMObject(GObject.Object):
|
||||
exit_build = self.build_process.proc.exitcode
|
||||
exitc = exit_vm or exit_build
|
||||
if not self.is_running() and exitc != 0:
|
||||
self.switch.handler_block(self.switch_handler_id)
|
||||
self.switch.set_active(False)
|
||||
self.switch.handler_unblock(self.switch_handler_id)
|
||||
with self.switch.handler_block(self.switch_handler_id):
|
||||
self.switch.set_active(False)
|
||||
log.error(f"VM exited with error. Exitcode: {exitc}")
|
||||
|
||||
def _on_switch_toggle(self, switch: Gtk.Switch, user_state: bool) -> None:
|
||||
|
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
from typing import Any, ClassVar
|
||||
from typing import Any, ClassVar, cast
|
||||
|
||||
import gi
|
||||
from clan_cli.clan_uri import ClanURI
|
||||
@ -31,8 +31,8 @@ class JoinValue(GObject.Object):
|
||||
|
||||
def __init__(self, url: ClanURI) -> None:
|
||||
super().__init__()
|
||||
self.url = url
|
||||
self.entry = None
|
||||
self.url: ClanURI = url
|
||||
self.entry: HistoryEntry | None = None
|
||||
|
||||
def __join(self) -> None:
|
||||
new_entry = add_history(self.url)
|
||||
@ -84,7 +84,7 @@ class JoinList:
|
||||
|
||||
value = JoinValue(uri)
|
||||
if value.url.machine.get_id() in [
|
||||
item.url.machine.get_id() for item in self.list_store
|
||||
cast(JoinValue, item).url.machine.get_id() for item in self.list_store
|
||||
]:
|
||||
log.info(f"Join request already exists: {value.url}. Ignoring.")
|
||||
return
|
||||
|
@ -1,16 +1,19 @@
|
||||
import os
|
||||
from collections.abc import Callable
|
||||
from functools import partial
|
||||
from typing import Any, Literal
|
||||
from typing import Any, Literal, TypeVar
|
||||
|
||||
import gi
|
||||
|
||||
gi.require_version("Adw", "1")
|
||||
from gi.repository import Adw, Gio, GObject, Gtk
|
||||
|
||||
# Define a TypeVar that is bound to GObject.Object
|
||||
ListItem = TypeVar("ListItem", bound=GObject.Object)
|
||||
|
||||
|
||||
def create_details_list(
|
||||
model: Gio.ListStore, render_row: Callable[[Gtk.ListBox, GObject], Gtk.Widget]
|
||||
model: Gio.ListStore, render_row: Callable[[Gtk.ListBox, ListItem], Gtk.Widget]
|
||||
) -> Gtk.ListBox:
|
||||
boxed_list = Gtk.ListBox()
|
||||
boxed_list.set_selection_mode(Gtk.SelectionMode.NONE)
|
||||
@ -49,7 +52,10 @@ class Details(Gtk.Box):
|
||||
def render_entry_row(
|
||||
self, boxed_list: Gtk.ListBox, item: PreferencesValue
|
||||
) -> Gtk.Widget:
|
||||
row = Adw.SpinRow.new_with_range(0, os.cpu_count(), 1)
|
||||
cores: int | None = os.cpu_count()
|
||||
fcores = float(cores) if cores else 1.0
|
||||
|
||||
row = Adw.SpinRow.new_with_range(0, fcores, 1)
|
||||
row.set_value(item.data)
|
||||
|
||||
return row
|
||||
|
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from functools import partial
|
||||
from typing import Any
|
||||
from typing import Any, TypeVar
|
||||
|
||||
import gi
|
||||
from clan_cli import history
|
||||
@ -17,9 +17,13 @@ from gi.repository import Adw, Gdk, Gio, GLib, GObject, Gtk
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
ListItem = TypeVar("ListItem", bound=GObject.Object)
|
||||
CustomStore = TypeVar("CustomStore", bound=Gio.ListModel)
|
||||
|
||||
|
||||
def create_boxed_list(
|
||||
model: Gio.ListStore, render_row: Callable[[Gtk.ListBox, GObject], Gtk.Widget]
|
||||
model: CustomStore,
|
||||
render_row: Callable[[Gtk.ListBox, ListItem], Gtk.Widget],
|
||||
) -> Gtk.ListBox:
|
||||
boxed_list = Gtk.ListBox()
|
||||
boxed_list.set_selection_mode(Gtk.SelectionMode.NONE)
|
||||
@ -47,8 +51,9 @@ class ClanList(Gtk.Box):
|
||||
def __init__(self, config: ClanConfig) -> None:
|
||||
super().__init__(orientation=Gtk.Orientation.VERTICAL)
|
||||
|
||||
self.app = Gio.Application.get_default()
|
||||
self.app.connect("join_request", self.on_join_request)
|
||||
app = Gio.Application.get_default()
|
||||
assert app is not None
|
||||
app.connect("join_request", self.on_join_request)
|
||||
|
||||
self.log_label: Gtk.Label = Gtk.Label()
|
||||
self.__init_machines = history.add.list_history()
|
||||
@ -78,6 +83,7 @@ class ClanList(Gtk.Box):
|
||||
add_action = Gio.SimpleAction.new("add", GLib.VariantType.new("s"))
|
||||
add_action.connect("activate", self.on_add)
|
||||
app = Gio.Application.get_default()
|
||||
assert app is not None
|
||||
app.add_action(add_action)
|
||||
|
||||
menu_model = Gio.Menu()
|
||||
@ -158,6 +164,7 @@ class ClanList(Gtk.Box):
|
||||
open_action = Gio.SimpleAction.new("edit", GLib.VariantType.new("s"))
|
||||
open_action.connect("activate", self.on_edit)
|
||||
app = Gio.Application.get_default()
|
||||
assert app is not None
|
||||
app.add_action(open_action)
|
||||
menu_model = Gio.Menu()
|
||||
menu_model.append("Edit", f"app.edit::{vm.get_id()}")
|
||||
@ -199,6 +206,7 @@ class ClanList(Gtk.Box):
|
||||
# Can't do this here because clan store is empty at this point
|
||||
if vm is not None:
|
||||
sub = row.get_subtitle()
|
||||
assert sub is not None
|
||||
row.set_subtitle(
|
||||
sub + "\nClan already exists. Joining again will update it"
|
||||
)
|
||||
|
@ -32,6 +32,7 @@ class MainWindow(Adw.ApplicationWindow):
|
||||
view.add_top_bar(header)
|
||||
|
||||
app = Gio.Application.get_default()
|
||||
assert app is not None
|
||||
self.tray_icon: TrayIcon = TrayIcon(app)
|
||||
|
||||
# Initialize all ClanStore
|
||||
|
@ -6,6 +6,7 @@
|
||||
, wrapGAppsHook
|
||||
, gtk4
|
||||
, gnome
|
||||
, pygobject-stubs
|
||||
, gobject-introspection
|
||||
, clan-cli
|
||||
, makeDesktopItem
|
||||
@ -41,7 +42,9 @@ python3.pkgs.buildPythonApplication {
|
||||
];
|
||||
|
||||
buildInputs = [ gtk4 libadwaita gnome.adwaita-icon-theme ];
|
||||
propagatedBuildInputs = [ pygobject3 clan-cli ];
|
||||
|
||||
# We need to propagate the build inputs to nix fmt / treefmt
|
||||
propagatedBuildInputs = [ pygobject3 clan-cli pygobject-stubs ];
|
||||
|
||||
# also re-expose dependencies so we test them in CI
|
||||
passthru = {
|
||||
|
@ -22,10 +22,6 @@ disallow_untyped_calls = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
|
||||
[[tool.mypy.overrides]]
|
||||
module = "gi.*"
|
||||
ignore_missing_imports = true
|
||||
|
||||
[[tool.mypy.overrides]]
|
||||
module = "clan_cli.*"
|
||||
ignore_missing_imports = true
|
||||
|
@ -24,7 +24,7 @@ mkShell (
|
||||
python3Packages.ipdb
|
||||
gtk4.dev
|
||||
libadwaita.devdoc # has the demo called 'adwaita-1-demo'
|
||||
] ++ clan-vm-manager.nativeBuildInputs;
|
||||
] ++ clan-vm-manager.nativeBuildInputs ++ clan-vm-manager.propagatedBuildInputs;
|
||||
|
||||
PYTHONBREAKPOINT = "ipdb.set_trace";
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user