psygnal.containers #

def clear(self) -> None:
    """Remove all items from the list.

    Overrides `MutableSequence.clear` (which pops one at a time) so the whole
    list is removed as a single contiguous block (one `batch_removing`/
    `batch_removed` pair). Items still go through `__delitem__`, so the per-item
    `removing`/`removed` events fire for each, highest index first, as before.
    """
    if self._data:
        del self[:]

def copy(self) -> Self:
    """Return a shallow copy of the list."""
    return self.__newlike__(self)

def extend(self, values: Iterable[_T]) -> None:
    """Extend list by appending all items from `values`.

    Overrides `MutableSequence.extend` (which appends one at a time) so the whole
    batch is bracketed by a single `batch_inserting`/`batch_inserted` pair. Items
    still go through `insert` one by one, emitting the per-item events.
    """
    values = list(values)
    if not values:
        return
    start, stop = len(self), len(self) + len(values)
    if self.events.batch_inserting:
        self.events.batch_inserting.emit(start, stop)
    self._batch_depth += 1
    try:
        for value in values:
            self.append(value)
    finally:
        self._batch_depth -= 1
    if self.events.batch_inserted:
        self.events.batch_inserted.emit(start, stop, values)

def insert(self, index: int, value: _T) -> None:
    """Insert `value` before index."""
    # `_pre_insert` may reject the value; run it before emitting anything so a
    # rejected value never leaves an unterminated batch_inserting behind.
    _value = self._pre_insert(value)
    if self._batch_depth:
        # inside extend(), which brackets the whole batch itself
        self._insert_one(index, value, _value)
        return
    # normalize for the (range-aware) batch_* signals; the per-item events
    # keep emitting the raw `index` exactly as before.
    norm = max(0, len(self) + index) if index < 0 else min(index, len(self))
    if self.events.batch_inserting:
        self.events.batch_inserting.emit(norm, norm + 1)
    self._insert_one(index, value, _value)
    if self.events.batch_inserted:
        self.events.batch_inserted.emit(norm, norm + 1, [value])

def move(self, src_index: int, dest_index: int = 0) -> bool:
    """Insert object at `src_index` before `dest_index`.

    Both indices refer to the list prior to any object removal
    (pre-move space).
    """
    if dest_index < 0:
        dest_index += len(self) + 1
    if dest_index in (src_index, src_index + 1):
        # this is a no-op
        return False

    self.events.moving.emit(src_index, dest_index)
    item = self._data.pop(src_index)
    if dest_index > src_index:
        dest_index -= 1
    self._data.insert(dest_index, item)
    self.events.moved.emit(src_index, dest_index, item)
    self.events.reordered.emit()
    return True

def move_multiple(self, sources: Iterable[Index], dest_index: int = 0) -> int:
    """Move a batch of `sources` indices, to a single destination.

    Note, if `dest_index` is higher than any of the `sources`, then
    the resulting position of the moved objects after the move operation
    is complete will be lower than `dest_index`.

    Parameters
    ----------
    sources : Iterable[Union[int, slice]]
        A sequence of indices
    dest_index : int, optional
        The destination index.  All sources will be inserted before this
        index (in pre-move space), by default 0... which has the effect of
        "bringing to front" everything in `sources`, or acting as a
        "reorder" method if `sources` contains all indices.

    Returns
    -------
    int
        The number of successful move operations completed.

    Raises
    ------
    TypeError
        If the destination index is a slice, or any of the source indices
        are not `int` or `slice`.
    """
    # calling list here makes sure that there are no index errors up front
    move_plan = list(self._move_plan(sources, dest_index))

    # don't assume index adjacency ... so move objects one at a time
    # this *could* be simplified with an intermediate list ... but this way
    # allows any views (such as QtViews) to update themselves more easily.
    # If this needs to be changed in the future for performance reasons,
    # then the associated QtListView will need to changed from using
    # `beginMoveRows` & `endMoveRows` to using `layoutAboutToBeChanged` &
    # `layoutChanged` while *manually* updating model indices with
    # `changePersistentIndexList`.  That becomes much harder to do with
    # nested tree-like models.
    with self.events.reordered.blocked():
        for src, dest in move_plan:
            self.move(src, dest)

    self.events.reordered.emit()
    return len(move_plan)

def reverse(self, *, emit_individual_events: bool = False) -> None:
    """Reverse list *IN PLACE*.

    Parameters
    ----------
    emit_individual_events : bool
        If True, the list is reversed one item at a time, emitting per-item
        events; otherwise the underlying data is reversed directly, by
        default False
    """
    if emit_individual_events:
        super().reverse()
    else:
        self._data.reverse()
    self.events.reordered.emit()

def clear(self) -> None:
    """Remove all elements from this set."""
    with self.events.items_changed.paused(_reduce_events):
        super().clear()

def difference_update(self, *s: Iterable[_T]) -> None:
    """Remove all elements of another set from this set."""
    with self.events.items_changed.paused(_reduce_events):
        super().difference_update(*s)

def intersection_update(self, *s: Iterable[_T]) -> None:
    """Update this set with the intersection of itself and another."""
    with self.events.items_changed.paused(_reduce_events):
        super().intersection_update(*s)

def symmetric_difference_update(self, __s: Iterable[_T]) -> None:
    """Update this set with the symmetric difference of itself and another.

    This will remove any items in this set that are also in `other`, and
    add any items in others that are not present in this set.
    """
    with self.events.items_changed.paused(_reduce_events, ((), ())):
        super().symmetric_difference_update(__s)

def update(self, *others: Iterable[_T]) -> None:
    """Update this set with the union of this set and others."""
    with self.events.items_changed.paused(_reduce_events):
        super().update(*others)

def deselect_all(self) -> None:
    """Deselect all items in the list."""
    self.selection.clear()

def insert(self, index: int, value: _T) -> None:
    """Insert item(s) into the list and update the selection.

    Parameters
    ----------
    index : int
        The index at which to insert the item.
    value : _T
        The item to insert.
    """
    super().insert(index, value)
    if self._activate_on_insert:
        self.selection.active = value

def remove_selected(self) -> tuple[_T, ...]:
    """Remove selected items from the list and the selection.

    Returns
    -------
    Tuple[_T, ...]
        The items that were removed.
    """
    selected_items = tuple(self.selection)
    idx = 0
    for item in list(self.selection):
        idx = self.index(item)
        self.remove(item)
    new_idx = max(0, idx - 1)
    if len(self) > new_idx:
        self.selection.add(self[new_idx])
    return selected_items

def select_all(self) -> None:
    """Select all items in the list."""
    self.selection.update(self)

def select_next(
    self, step: int = 1, expand_selection: bool = False, wraparound: bool = False
) -> None:
    """Select the next item in the list.

    Parameters
    ----------
    step : int
        The step size to take when picking the next item, by default 1
    expand_selection : bool
        If True, will expand the selection to contain the both the current item and
        the next item, by default False
    wraparound : bool
        Whether to return to the beginning of the list of the end has been reached,
        by default False
    """
    if len(self) == 0:
        return
    elif not self.selection:
        idx = -1 if step > 0 else 0
    else:
        idx = self.index(self.selection._current) + step
    idx_in_sequence = len(self) > idx >= 0
    if wraparound:
        idx = idx % len(self)
    elif not idx_in_sequence:
        idx = -1 if step > 0 else 0
    next_item = self[idx]
    if expand_selection:
        self.selection.add(next_item)
        self.selection._current = next_item
    else:
        self.selection.active = next_item

def select_previous(
    self, expand_selection: bool = False, wraparound: bool = False
) -> None:
    """Select the previous item in the list.

    Parameters
    ----------
    expand_selection : bool
        If True, will expand the selection to contain both the current item and
        the previous item, by default False
    wraparound : bool
        Whether to return to the end of the list if the beginning has been
        reached, by default False
    """
    self.select_next(
        step=-1, expand_selection=expand_selection, wraparound=wraparound
    )

def clear(self, keep_current: bool = False) -> None:
    """Clear the selection.

    Parameters
    ----------
    keep_current : bool
        If `False` (the default), the "current" item will also be set to None.
    """
    if not keep_current:
        self._current = None
    super().clear()

def replace_selection(self, new_selection: Iterable[_T]) -> None:
    """Replace the current selection with `new_selection`.

    This is equivalent to calling `intersection_update` followed by `update`,
    but is more efficient because it only emits a single `items_changed` event.
    """
    with self.events.items_changed.paused(_reduce_events):
        self.intersection_update(new_selection)
        self.update(new_selection)

def select_only(self, obj: _T) -> None:
    """Unselect everything but `obj`. Add to selection if not currently selected."""
    with self.events.items_changed.paused(_reduce_events):
        self.intersection_update({obj})
        self.add(obj)

def toggle(self, obj: _T) -> None:
    """Toggle selection state of obj."""
    self.symmetric_difference_update({obj})