358 lines
11 KiB
Python
358 lines
11 KiB
Python
import logging
|
|
from typing import (
|
|
Any, Callable, Dict, Iterable, List, Mapping, MutableSequence, Optional,
|
|
Sequence, Tuple, Union
|
|
)
|
|
|
|
from PyQt5.QtCore import (
|
|
QAbstractListModel, QModelIndex, QObject, Qt, pyqtProperty, pyqtSignal,
|
|
pyqtSlot
|
|
)
|
|
|
|
from .list_item import ListItem
|
|
|
|
Index = Union[int, str]
|
|
NewItem = Union[ListItem, Mapping[str, Any], Sequence]
|
|
|
|
|
|
class _GetFail:
|
|
pass
|
|
|
|
|
|
class _PopFail:
|
|
pass
|
|
|
|
|
|
class ListModel(QAbstractListModel):
|
|
rolesSet = pyqtSignal()
|
|
changed = pyqtSignal()
|
|
countChanged = pyqtSignal(int)
|
|
|
|
def __init__(self,
|
|
initial_data: Optional[List[NewItem]] = None,
|
|
container: Callable[..., MutableSequence] = list,
|
|
parent: QObject = None) -> None:
|
|
super().__init__(parent)
|
|
self._data: MutableSequence[ListItem] = container()
|
|
|
|
if initial_data:
|
|
self.extend(initial_data)
|
|
|
|
|
|
def __repr__(self) -> str:
|
|
return "%s(%r)" % (type(self).__name__, self._data)
|
|
|
|
|
|
def __contains__(self, index: Index) -> bool:
|
|
if isinstance(index, str):
|
|
try:
|
|
self.indexWhere(self.mainKey, index)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
return index in self._data
|
|
|
|
|
|
def __getitem__(self, index: Index) -> ListItem:
|
|
return self.get(index)
|
|
|
|
|
|
def __setitem__(self, index: Index, value: NewItem) -> None:
|
|
self.set(index, value)
|
|
|
|
|
|
def __delitem__(self, index: Index) -> None:
|
|
self.remove(index)
|
|
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._data)
|
|
|
|
|
|
def __iter__(self) -> Iterable[NewItem]:
|
|
return iter(self._data)
|
|
|
|
|
|
@pyqtSlot(result=str)
|
|
def repr(self) -> str:
|
|
return self.__repr__()
|
|
|
|
|
|
@pyqtProperty("QStringList", notify=rolesSet)
|
|
def roles(self) -> Tuple[str, ...]:
|
|
return self._data[0].roles if self._data else () # type: ignore
|
|
|
|
|
|
@pyqtProperty("QVariant", notify=rolesSet)
|
|
def mainKey(self) -> Optional[str]:
|
|
return self._data[0].mainKey if self._data else None
|
|
|
|
|
|
def roleNumbers(self) -> Dict[str, int]:
|
|
return {name: Qt.UserRole + i
|
|
for i, name in enumerate(self.roles, 1)} \
|
|
if self._data else {}
|
|
|
|
|
|
def roleNames(self) -> Dict[int, bytes]:
|
|
return {Qt.UserRole + i: bytes(name, "utf-8")
|
|
for i, name in enumerate(self.roles, 1)} \
|
|
if self._data else {}
|
|
|
|
|
|
def data(self, index: QModelIndex, role: int = Qt.DisplayRole) -> Any:
|
|
if role <= Qt.UserRole:
|
|
return None
|
|
|
|
return getattr(self._data[index.row()],
|
|
str(self.roleNames()[role], "utf8"))
|
|
|
|
|
|
def rowCount(self, _: QModelIndex = QModelIndex()) -> int:
|
|
return len(self)
|
|
|
|
|
|
def _convert_new_value(self, value: NewItem) -> ListItem:
|
|
def convert() -> ListItem:
|
|
if self._data and isinstance(value, Mapping):
|
|
if not set(value.keys()) <= set(self.roles):
|
|
raise ValueError(
|
|
f"{value}: must have all these keys: {self.roles}"
|
|
)
|
|
|
|
return type(self._data[0])(**value)
|
|
|
|
if not self._data and isinstance(value, Mapping):
|
|
raise NotImplementedError("First item must be set from Python")
|
|
|
|
if self._data and isinstance(value, type(self._data[0])):
|
|
return value
|
|
|
|
if not self._data and isinstance(value, ListItem):
|
|
return value
|
|
|
|
raise TypeError("%r: must be mapping or %s" % (
|
|
value,
|
|
type(self._data[0]).__name__ if self._data else "ListItem"
|
|
))
|
|
|
|
value = convert()
|
|
value.setParent(self)
|
|
return value
|
|
|
|
|
|
@pyqtProperty(int, notify=countChanged)
|
|
def count(self) -> int:
|
|
return len(self)
|
|
|
|
|
|
@pyqtSlot(str, "QVariant", result=int)
|
|
def indexWhere(self, prop: str, is_value: Any) -> int:
|
|
for i, item in enumerate(self._data):
|
|
if getattr(item, prop) == is_value:
|
|
return i
|
|
|
|
raise ValueError(f"No item in model data with "
|
|
f"property {prop!r} set to {is_value!r}.")
|
|
|
|
|
|
@pyqtSlot(int, result="QVariant")
|
|
@pyqtSlot(str, result="QVariant")
|
|
@pyqtSlot(int, "QVariant", result="QVariant")
|
|
@pyqtSlot(str, "QVariant", result="QVariant")
|
|
def get(self, index: Index, default: Any = _GetFail()) -> ListItem:
|
|
try:
|
|
i_index: int = self.indexWhere(self.mainKey, index) \
|
|
if isinstance(index, str) else index
|
|
|
|
return self._data[i_index]
|
|
|
|
except (ValueError, IndexError):
|
|
if isinstance(default, _GetFail):
|
|
raise
|
|
return default
|
|
|
|
|
|
@pyqtSlot(int, "QVariantMap")
|
|
def insert(self, index: int, value: NewItem) -> None:
|
|
value = self._convert_new_value(value)
|
|
|
|
self.beginInsertRows(QModelIndex(), index, index)
|
|
|
|
had_data = bool(self._data)
|
|
self._data.insert(index, value)
|
|
if not had_data:
|
|
self.rolesSet.emit()
|
|
|
|
self.endInsertRows()
|
|
|
|
self.countChanged.emit(len(self))
|
|
self.changed.emit()
|
|
|
|
|
|
@pyqtSlot("QVariantMap")
|
|
def append(self, value: NewItem) -> None:
|
|
self.insert(len(self), value)
|
|
|
|
|
|
@pyqtSlot(list)
|
|
def extend(self, values: Iterable[NewItem]) -> None:
|
|
for val in values:
|
|
self.append(val)
|
|
|
|
|
|
@pyqtSlot(int, "QVariantMap", result=int)
|
|
@pyqtSlot(int, "QVariantMap", "QStringList", result=int)
|
|
@pyqtSlot(str, "QVariantMap", result=int)
|
|
@pyqtSlot(str, "QVariantMap", "QStringList", result=int)
|
|
def update(self,
|
|
index: Index,
|
|
value: NewItem,
|
|
ignore_roles: Sequence[str] = ()) -> int:
|
|
value = self._convert_new_value(value)
|
|
|
|
i_index: int = self.indexWhere(self.mainKey, index) \
|
|
if isinstance(index, str) else index
|
|
|
|
to_update = self[i_index]
|
|
|
|
for role in self.roles:
|
|
if role not in ignore_roles:
|
|
try:
|
|
setattr(to_update, role, getattr(value, role))
|
|
except AttributeError: # constant/not settable
|
|
pass
|
|
|
|
qidx = QAbstractListModel.index(self, i_index, 0)
|
|
updated = [number for name, number in self.roleNumbers().items()
|
|
if name not in ignore_roles]
|
|
self.dataChanged.emit(qidx, qidx, updated)
|
|
self.changed.emit()
|
|
return i_index
|
|
|
|
|
|
@pyqtSlot(str, "QVariantMap")
|
|
@pyqtSlot(str, "QVariantMap", int)
|
|
@pyqtSlot(str, "QVariantMap", int, int)
|
|
@pyqtSlot(str, "QVariantMap", int, int, "QStringList")
|
|
def upsert(self,
|
|
where_main_key_is: Any,
|
|
update_with: NewItem,
|
|
new_index_if_insert: Optional[int] = None,
|
|
new_index_if_update: Optional[int] = None,
|
|
ignore_roles: Sequence[str] = ()) -> None:
|
|
try:
|
|
index = self.update(where_main_key_is, update_with, ignore_roles)
|
|
except (IndexError, ValueError):
|
|
self.insert(new_index_if_insert or len(self), update_with)
|
|
else:
|
|
if new_index_if_update:
|
|
self.move(index, new_index_if_update)
|
|
|
|
|
|
@pyqtSlot(int, list)
|
|
@pyqtSlot(str, list)
|
|
def set(self, index: Index, value: NewItem) -> None:
|
|
i_index: int = self.indexWhere(self.mainKey, index) \
|
|
if isinstance(index, str) else index
|
|
|
|
qidx = QAbstractListModel.index(self, i_index, 0)
|
|
value = self._convert_new_value(value)
|
|
self._data[i_index] = value
|
|
self.dataChanged.emit(qidx, qidx, self.roleNames())
|
|
self.changed.emit()
|
|
|
|
|
|
@pyqtSlot(int, str, "QVariant")
|
|
@pyqtSlot(str, str, "QVariant")
|
|
def setProperty(self, index: Index, prop: str, value: Any) -> None:
|
|
i_index: int = self.indexWhere(self.mainKey, index) \
|
|
if isinstance(index, str) else index
|
|
|
|
setattr(self[i_index], prop, value)
|
|
qidx = QAbstractListModel.index(self, i_index, 0)
|
|
self.dataChanged.emit(qidx, qidx, (self.roleNumbers()[prop],))
|
|
self.changed.emit()
|
|
|
|
|
|
@pyqtSlot(int, int)
|
|
@pyqtSlot(int, int, int)
|
|
@pyqtSlot(str, int)
|
|
@pyqtSlot(str, int, int)
|
|
def move(self, from_: Index, to: int, n: int = 1) -> None:
|
|
# pylint: disable=invalid-name
|
|
i_from: int = self.indexWhere(self.mainKey, from_) \
|
|
if isinstance(from_, str) else from_
|
|
|
|
qlast = i_from + n - 1
|
|
|
|
if (n <= 0) or (i_from == to) or (qlast == to) or \
|
|
not (len(self) > qlast >= 0) or \
|
|
not len(self) >= to >= 0:
|
|
return
|
|
|
|
qidx = QModelIndex()
|
|
qto = min(len(self), to + n if to > i_from else to)
|
|
# print(f"self.beginMoveRows(qidx, {i_from}, {qlast}, qidx, {qto})")
|
|
valid = self.beginMoveRows(qidx, i_from, qlast, qidx, qto)
|
|
|
|
if not valid:
|
|
logging.warning("Invalid move operation - %r", locals())
|
|
return
|
|
|
|
last = i_from + n
|
|
cut = self._data[i_from:last]
|
|
del self._data[i_from:last]
|
|
self._data[to:to] = cut
|
|
|
|
self.endMoveRows()
|
|
self.changed.emit()
|
|
|
|
|
|
@pyqtSlot(int)
|
|
@pyqtSlot(str)
|
|
def remove(self, index: Index) -> None:
|
|
i_index: int = self.indexWhere(self.mainKey, index) \
|
|
if isinstance(index, str) else index
|
|
|
|
self.beginRemoveRows(QModelIndex(), i_index, i_index)
|
|
del self._data[i_index]
|
|
self.endRemoveRows()
|
|
|
|
self.countChanged.emit(len(self))
|
|
self.changed.emit()
|
|
|
|
|
|
@pyqtSlot(int, result="QVariant")
|
|
@pyqtSlot(str, result="QVariant")
|
|
def pop(self, index: Index, default: Any = _PopFail()) -> ListItem:
|
|
try:
|
|
i_index: int = self.indexWhere(self.mainKey, index) \
|
|
if isinstance(index, str) else index
|
|
item = self[i_index]
|
|
|
|
except (ValueError, IndexError):
|
|
if isinstance(default, _PopFail):
|
|
raise
|
|
return default
|
|
|
|
self.beginRemoveRows(QModelIndex(), i_index, i_index)
|
|
del self._data[i_index]
|
|
self.endRemoveRows()
|
|
|
|
self.countChanged.emit(len(self))
|
|
self.changed.emit()
|
|
return item
|
|
|
|
|
|
@pyqtSlot()
|
|
def clear(self) -> None:
|
|
# Reimplemented for performance reasons (begin/endRemoveRows)
|
|
self.beginRemoveRows(QModelIndex(), 0, len(self))
|
|
self._data.clear()
|
|
self.endRemoveRows()
|
|
|
|
self.countChanged.emit(len(self))
|
|
self.changed.emit()
|