Windows support

This commit is contained in:
sarah committed 2026-10-09 14:30:59 +02:00
1 parent 6b2b35f5da
commit b295db3fb6
9 files changed
+136 -26

No files matched your search

+51 -4
View File
@@ -17,15 +17,23 @@ Falls back to polling /run/media/<user> if udisks2 isn't reachable.
Everything runs on the page's own event loop (Page.run_task), so calling
.update() from here is safe."""
from __future__ import annotations # the D-Bus type hints below don't exist on Windows
import asyncio
import getpass
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional
import flet as ft
from dbus_next import BusType, Message, MessageType, Variant
from dbus_next.aio import MessageBus
IsWindows = sys.platform == "win32"
if not IsWindows:
# udisks2 is Linux-only; Windows lists its drives itself (ListWindowsDrives).
from dbus_next import BusType, Message, MessageType, Variant
from dbus_next.aio import MessageBus
from Core import AppConfig, NotificationStyle, Palette
@@ -35,7 +43,7 @@ BlockInterface = f"{UDisks}.Block"
FilesystemInterface = f"{UDisks}.Filesystem"
DriveInterface = f"{UDisks}.Drive"
PollInterval = 3.0 # seconds, only for the no-udisks2 fallback
PollInterval = 3.0 # seconds, for Windows and the no-udisks2 fallback
MountRoots = [Path("/run/media") / getpass.getuser(), Path("/media") / getpass.getuser()]
@@ -71,6 +79,39 @@ def _UserMount(MountPoint: Path) -> bool:
return any(Root in MountPoint.parents for Root in MountRoots)
def ListWindowsDrives(ShowInternal: bool, ShowSystemDrive: bool) -> list[Device]:
"""Windows: removable drives (USB sticks, memory cards, discs with a disc
in), plus fixed and network drives with ShowInternal, and the system drive
(C:) with ShowSystemDrive. Windows mounts drives itself, so there's nothing
to mount; ejecting is left to Windows' own "Safely remove"."""
import ctypes
Kernel = ctypes.windll.kernel32
Removable, Fixed, Remote, CdRom = 2, 3, 4, 5
SystemDrive = (os.environ.get("SystemDrive") or "C:").upper().rstrip("\\") + "\\"
Devices: list[Device] = []
Mask = Kernel.GetLogicalDrives()
for Index in range(26):
if not Mask & (1 << Index):
continue
Root = f"{chr(65 + Index)}:\\"
Kind = Kernel.GetDriveTypeW(Root)
IsSystem = Root.upper() == SystemDrive
if IsSystem and not ShowSystemDrive:
continue
if Kind in (Fixed, Remote) and not (ShowInternal or IsSystem):
continue
if Kind not in (Removable, Fixed, Remote, CdRom):
continue
Name = ctypes.create_unicode_buffer(261)
# Fails for an empty card reader or disc drive: nothing to show then.
if not Kernel.GetVolumeInformationW(Root, Name, 261, None, None, None, None, 0):
continue
Fallback = {Removable: "Removable drive", CdRom: "Disc", Remote: "Network drive"}.get(Kind, "Local disk")
Devices.append(Device(Root, f"{Name.value or Fallback} ({Root[:2]})", Path(Root), IsSystemMount=IsSystem))
return Devices
def ListFallbackDevices() -> list[Device]:
Devices: list[Device] = []
for Root in MountRoots:
@@ -210,6 +251,9 @@ class DevicesPanel(ft.Container):
self.Client = None
async def _Start(self) -> None:
if IsWindows:
await self._PollLoop() # no udisks2: Windows' own drive list, checked every few seconds
return
try:
self.Client = await UDisksClient.Connect()
await self.Client.Watch(self._ScheduleRefresh)
@@ -221,7 +265,10 @@ class DevicesPanel(ft.Container):
async def _PollLoop(self) -> None:
while not self.Unmounted:
self._Show(ListFallbackDevices())
if IsWindows:
self._Show(ListWindowsDrives(self.Config.ShowInternalDrives, self.Config.ShowSystemDrive))
else:
self._Show(ListFallbackDevices())
await asyncio.sleep(PollInterval)
def _ScheduleRefresh(self) -> None: