from __future__ import annotations import os from enum import Enum, auto from pathlib import Path from typing import Optional, overload from gi.repository import Gio, GObject from .reactive import Ref, Watcher class FileType(Enum): DIRECTORY = auto() VIDEO = auto() class FileItem(GObject.Object, Watcher): file_type: FileType full_path: Path thumbnail: Ref[bytes] attempted_thumbnail: Ref[bool] saved_position: Ref[int] saved_duration: Ref[int] saved_subtitle_track: Ref[int] saved_audio_track: Ref[int] __gtype_name__ = "FileItem" def __init__(self, name: str, file_type: FileType, full_path: Path): super().__init__() self.name = name self.file_type = file_type self.full_path = full_path self.thumbnail = Ref(b"") self.attempted_thumbnail = Ref(False) self.saved_position = Ref(self._load_attribute("position", 0)) self.saved_duration = Ref(self._load_attribute("duration", 1)) self.saved_subtitle_track = Ref(self._load_attribute("subtitle_track", -2)) self.saved_audio_track = Ref(self._load_attribute("audio_track", 0)) self.watch_all() def _watch_saved_position(self): saved_position = self.saved_position.value self._save_attribute("position", saved_position if saved_position > 0 else None) def _watch_saved_duration(self): saved_duration = self.saved_duration.value self._save_attribute("duration", saved_duration if saved_duration > 0 else None) def _watch_saved_subtitle_track(self): saved_subtitle_track = self.saved_subtitle_track.value self._save_attribute( "subtitle_track", saved_subtitle_track if saved_subtitle_track > -2 else None, ) def _watch_saved_audio_track(self): saved_audio_track = self.saved_audio_track.value self._save_attribute( "audio_track", saved_audio_track if saved_audio_track > -1 else None, ) @overload def _load_attribute(self, name: str, dfl: str) -> str: ... @overload def _load_attribute(self, name: str, dfl: bytes) -> bytes: ... @overload def _load_attribute(self, name: str, dfl: int) -> int: ... def _load_attribute(self, name: str, dfl: str | bytes | int) -> str | bytes | int: try: strval = os.getxattr(self.full_path, f"user.lazy_player.{name}") return type(dfl)(strval) except OSError: return dfl def _save_attribute(self, name: str, value: str | bytes | float | int | None) -> None: try: if value is None: os.removexattr(self.full_path, f"user.lazy_player.{name}") else: if isinstance(value, bytes): os.setxattr(self.full_path, f"user.lazy_player.{name}", value) else: os.setxattr( self.full_path, f"user.lazy_player.{name}", str(value).encode("utf8") ) except OSError: pass class FileListModel(GObject.Object, Gio.ListModel): """A ListModel implementation for FileItems""" __gtype_name__ = "FileListModel" items: list[FileItem] def __init__(self): super().__init__() self.items = [] def do_get_item_type(self) -> GObject.GType: return GObject.type_from_name("FileItem") def do_get_n_items(self) -> int: return len(self.items) def do_get_item(self, position: int) -> Optional[FileItem]: if 0 <= position < len(self.items): return self.items[position] return None def remove_all(self) -> None: removed = len(self.items) self.items = [] self.items_changed(0, removed, 0) def set_items(self, items: list[FileItem]): removed = len(self.items) self.items = list(items) self.items_changed(0, removed, len(self.items)) def append(self, item: FileItem) -> None: pos = len(self.items) self.items.append(item) self.items_changed(pos, 0, 1) def remove(self, position: int) -> None: if 0 <= position < len(self.items): self.items.pop(position) self.items_changed(position, 1, 0)