psygnal.containers #

Bases: ProxyEvents

Events emitted by EventedCallableObjectProxy.

Attributes:

• called –

  Emitted when the object is called.

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431

def __init__(self, instance: Any = None) -> None:
    cls = type(self)
    if not hasattr(cls, "_psygnal_signals"):
        raise TypeError(
            "Cannot instantiate `SignalGroup` directly.  Use a subclass instead."
        )

    self._psygnal_instances = {
        name: (
            sig._create_signal_instance(self)
            if name in cls._psygnal_name_conflicts
            else sig.__get__(self, cls)
        )
        for name, sig in cls._psygnal_signals.items()
    }
    self._psygnal_relay = SignalRelay(self._psygnal_instances, instance)

called = Signal(tuple, dict)

Bases: SignalGroup

Events available on EventedDict.

Attributes:

• added –

  (key, value) emitted after a value is added at key

• adding –

  (key,) emitted before an item is added at key

• changed –

  (key, old_value, new_value) emitted before old_value is replaced with

• changing –

  (key, old_value, new_value) emitted before old_value is replaced with

• removed –

  (key, value) emitted after value is removed at key

• removing –

  (key,) emitted before an item is removed at key

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431

def __init__(self, instance: Any = None) -> None:
    cls = type(self)
    if not hasattr(cls, "_psygnal_signals"):
        raise TypeError(
            "Cannot instantiate `SignalGroup` directly.  Use a subclass instead."
        )

    self._psygnal_instances = {
        name: (
            sig._create_signal_instance(self)
            if name in cls._psygnal_name_conflicts
            else sig.__get__(self, cls)
        )
        for name, sig in cls._psygnal_signals.items()
    }
    self._psygnal_relay = SignalRelay(self._psygnal_instances, instance)

added = DictSignal(object, object)

adding = DictSignal(object)

changed = DictSignal(object, object, object)

changing = DictSignal(object)

removed = DictSignal(object, object)

removing = DictSignal(object)

Bases: EventedObjectProxy

Create a proxy of target that includes an events psygnal.SignalGroup.

target must be callable.

Signals will be emitted whenever an attribute is set or deleted, or (if the object implements __getitem__) whenever an item is set or deleted. If the object supports in-place modification (i.e. any of the __i{}__ magic methods), then an in_place event is emitted (with the name of the method) whenever any of them are used. Lastly, if the item is called, a called event is emitted with the (args, kwargs) used in the call.

The events available at target.events include:

• attribute_set: Signal(str, object)
• attribute_deleted: Signal(str)
• item_set: Signal(object, object)
• item_deleted: Signal(object)
• in_place: Signal(str, object)
• called: Signal(tuple, dict)

Parameters:

• target #

  (Callable) –

  An callable object to wrap

Attributes:

• events (CallableProxyEvents) –

  SignalGroup containing events for this object proxy.

215
216

def __init__(self, target: Callable):
    super().__init__(target)

events: CallableProxyEvents

Bases: TypedMutableMapping[_K, _V]

Mutable mapping that emits events when altered.

This class is designed to behave exactly like the builtin dict, but will emit events before and after all mutations (addition, removal, and changing).

Parameters:

• data #

  (Union[Mapping[_K, _V], Iterable[Tuple[_K, _V]], None], default: None ) –

  Data suitable of passing to dict(). Mapping of {key: value} pairs, or Iterable of two-tuples [(key, value), ...], or None to create an

• basetype #

  (TypeOrSequenceOfTypes, default: () ) –

  Type or Sequence of Type objects. If provided, values entered into this Mapping must be an instance of one of the provided types. by default ().

Attributes:

• events (DictEvents) –

  The SignalGroup object that emits all events available on an EventedDict.

186
187
188
189
190
191
192
193
194

def __init__(
    self,
    data: DictArg | None = None,
    *,
    basetype: TypeOrSequenceOfTypes = (),
    **kwargs: _V,
):
    self.events = DictEvents()
    super().__init__(data, basetype=basetype, **kwargs)

Bases: MutableSequence[_T]

Mutable Sequence that emits events when altered.

This class is designed to behave exactly like the builtin list, but will emit events before and after all mutations (insertion, removal, setting, and moving).

Parameters:

• data #

  (iterable, default: () ) –

  Elements to initialize the list with.

• hashable #

  (bool, default: True ) –

  Whether the list should be hashable as id(self). By default True.

• child_events #

  (bool, default: True ) –

  Whether to re-emit events from emitted from evented items in the list (i.e. items that have SignalInstances). If True, child events can be connected at EventedList.events.child_event. By default, True.

Attributes:

• events (ListEvents) –

  SignalGroup that with events related to list mutation. (see ListEvents)

Methods:

• copy –

  Return a shallow copy of the list.

• insert –

  Insert value before index.

• move –

  Insert object at src_index before dest_index.

• move_multiple –

  Move a batch of sources indices, to a single destination.

• reverse –

  Reverse list IN PLACE.

122
123
124
125
126
127
128
129
130
131
132
133
134

def __init__(
    self,
    data: Iterable[_T] = (),
    *,
    hashable: bool = True,
    child_events: bool = True,
):
    super().__init__()
    self._data: list[_T] = []
    self._hashable = hashable
    self._child_events = child_events
    self.events = ListEvents(instance=self)
    self.extend(data)

copy() -> Self

223
224
225

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

insert(index: int, value: _T) -> None

143
144
145
146
147
148
149

def insert(self, index: int, value: _T) -> None:
    """Insert `value` before index."""
    _value = self._pre_insert(value)
    self.events.inserting.emit(index)
    self._data.insert(index, _value)
    self.events.inserted.emit(index, value)
    self._post_insert(value)

move(src_index: int, dest_index: int = 0) -> bool

277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296

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

move_multiple(
    sources: Iterable[Index], dest_index: int = 0
) -> int

298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343

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)

reverse(*, emit_individual_events: bool = False) -> None

269
270
271
272
273
274
275

def reverse(self, *, emit_individual_events: bool = False) -> None:
    """Reverse list *IN PLACE*."""
    if emit_individual_events:
        super().reverse()
    else:
        self._data.reverse()
    self.events.reordered.emit()

Bases: ObjectProxy, Generic[T]

Create a proxy of target that includes an events psygnal.SignalGroup.

Provides an "evented" subclasses of wrapt.ObjectProxy

Signals will be emitted whenever an attribute is set or deleted, or (if the object implements __getitem__) whenever an item is set or deleted. If the object supports in-place modification (i.e. any of the __i{}__ magic methods), then an in_place event is emitted (with the name of the method) whenever any of them are used.

The events available at target.events include:

• attribute_set: Signal(str, object)
• attribute_deleted: Signal(str)
• item_set: Signal(object, object)
• item_deleted: Signal(object)
• in_place: Signal(str, object)

Parameters:

• target #

  (Any) –

  An object to wrap

Attributes:

• events (ProxyEvents) –

  SignalGroup containing events for this object proxy.

89
90

def __init__(self, target: Any):
    super().__init__(target)

events: ProxyEvents

Bases: EventedSet, OrderedSet[_T]

A ordered variant of EventedSet that maintains insertion order.

Parameters:

• iterable #

  (Iterable[_T], default: () ) –

  Data to populate the set. If omitted, an empty set is created.

Attributes:

• events (SetEvents) –

  SignalGroup that with events related to set mutation. (see SetEvents)

361
362

def __init__(self, iterable: Iterable[_T] = ()):
    super().__init__(iterable)

Bases: _BaseMutableSet[_T]

A set with an items_changed signal that emits when items are added/removed.

Parameters:

• iterable #

  (Iterable[_T], default: () ) –

  Data to populate the set. If omitted, an empty set is created.

Attributes:

• events (SetEvents) –

  SignalGroup that with events related to set mutation. (see SetEvents)

Examples:

>>> from psygnal.containers import EventedSet
>>>
>>> my_set = EventedSet([1, 2, 3])
>>> my_set.events.items_changed.connect(
>>>     lambda a, r: print(f"added={a}, removed={r}")
>>> )
>>> my_set.update({3, 4, 5})
added=(4, 5), removed=()

Multi-item events will be reduced into a single emission:

>>> my_set.symmetric_difference_update({4, 5, 6, 7})
added=(6, 7), removed=(4, 5)

>>> my_set
EventedSet({1, 2, 3, 6, 7})

Methods:

• clear –

  Remove all elements from this set.

• difference_update –

  Remove all elements of another set from this set.

• intersection_update –

  Update this set with the intersection of itself and another.

• symmetric_difference_update –

  Update this set with the symmetric difference of itself and another.

• update –

  Update this set with the union of this set and others.

286
287
288

def __init__(self, iterable: Iterable[_T] = ()):
    self.events = self._get_events_class()
    super().__init__(iterable)

clear() -> None

295
296
297
298

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

difference_update(*s: Iterable[_T]) -> None

300
301
302
303

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)

intersection_update(*s: Iterable[_T]) -> None

305
306
307
308

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)

symmetric_difference_update(__s: Iterable[_T]) -> None

310
311
312
313
314
315
316
317

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)

update(*others: Iterable[_T]) -> None

290
291
292
293

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)

Bases: SignalGroup

Events available on EventedList.

Attributes:

• changed –

  (index_or_slice, old_value, value) emitted when index is set from

• child_event –

  (EmissionInfo) emitted when an object in the list emits an

• inserted –

  (index, value) emitted after value is inserted at index

• inserting –

  (index) emitted before an item is inserted at index

• moved –

  (index, new_index, value) emitted after value is moved from

• moving –

  (index, new_index) emitted before an item is moved from index to

• removed –

  (index, value) emitted after value is removed at index

• removing –

  (index) emitted before an item is removed at index

• reordered –

  Emitted when the list is reordered (eg. moved/reversed).

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431

def __init__(self, instance: Any = None) -> None:
    cls = type(self)
    if not hasattr(cls, "_psygnal_signals"):
        raise TypeError(
            "Cannot instantiate `SignalGroup` directly.  Use a subclass instead."
        )

    self._psygnal_instances = {
        name: (
            sig._create_signal_instance(self)
            if name in cls._psygnal_name_conflicts
            else sig.__get__(self, cls)
        )
        for name, sig in cls._psygnal_signals.items()
    }
    self._psygnal_relay = SignalRelay(self._psygnal_instances, instance)

changed = ListSignal(object, object, object)

child_event = Signal(EmissionInfo)

inserted = ListSignal(int, object)

inserting = ListSignal(int)

moved = ListSignal(int, int, object)

moving = ListSignal(int, int)

removed = ListSignal(int, object)

removing = ListSignal(int)

reordered = Signal()

Bases: _BaseMutableSet[_T]

A set that preserves insertion order, uses dict behind the scenes.

216
217
218

def __init__(self, iterable: Iterable[_T] = ()):
    self._data = {}
    self.update(iterable)

Bases: SignalGroup

Events emitted by EventedObjectProxy and EventedCallableObjectProxy.

Attributes:

• attribute_deleted –

  Emitted when an attribute is deleted.

• attribute_set –

  Emitted when an attribute is set.

• in_place –

  Emitted when an in-place operation is performed.

• item_deleted –

  Emitted when an item is deleted.

• item_set –

  Emitted when an item is set.

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431

def __init__(self, instance: Any = None) -> None:
    cls = type(self)
    if not hasattr(cls, "_psygnal_signals"):
        raise TypeError(
            "Cannot instantiate `SignalGroup` directly.  Use a subclass instead."
        )

    self._psygnal_instances = {
        name: (
            sig._create_signal_instance(self)
            if name in cls._psygnal_name_conflicts
            else sig.__get__(self, cls)
        )
        for name, sig in cls._psygnal_signals.items()
    }
    self._psygnal_relay = SignalRelay(self._psygnal_instances, instance)

attribute_deleted = Signal(str)

attribute_set = Signal(str, object)

in_place = Signal(str, object)

item_deleted = Signal(object)

item_set = Signal(object, object)

Bases: Selectable[_T], EventedList[_T]

EventedList subclass with a built in selection model.

In addition to all EventedList properties, this class also has a selection attribute that manages a set of selected items in the list.

Parameters:

• data #

  (iterable, default: () ) –

  Elements to initialize the list with.

• hashable #

  (bool, default: True ) –

  Whether the list should be hashable as id(self). By default True.

• child_events #

  (bool, default: False ) –

  Whether to re-emit events from emitted from evented items in the list (i.e. items that have SignalInstances). If True, child events can be connected at EventedList.events.child_event. By default, False.

Attributes:

• events (ListEvents) –

  SignalGroup that with events related to list mutation. (see ListEvents)

• selection (Selection) –

  An evented set containing the currently selected items, along with an active and current item. (See Selection)

Methods:

• deselect_all –

  Deselect all items in the list.

• insert –

  Insert item(s) into the list and update the selection.

• remove_selected –

  Remove selected items from the list and the selection.

• select_all –

  Select all items in the list.

• select_next –

  Select the next item in the list.

• select_previous –

  Select the previous item in the list.

40
41
42
43
44
45
46
47
48
49

def __init__(
    self,
    data: Iterable[_T] = (),
    *,
    hashable: bool = True,
    child_events: bool = False,
):
    self._activate_on_insert: bool = True
    super().__init__(data=data, hashable=hashable, child_events=child_events)
    self.events.removed.connect(self._on_item_removed)

deselect_all() -> None

64
65
66

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

insert(index: int, value: _T) -> None

54
55
56
57
58

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

remove_selected() -> tuple[_T, ...]

110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

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

select_all() -> None

60
61
62

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

select_next(
    step: int = 1,
    expand_selection: bool = False,
    wraparound: bool = False,
) -> None

 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

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

select_previous(
    expand_selection: bool = False, wraparound: bool = False
) -> None

102
103
104
105
106
107
108

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

Bases: EventedOrderedSet[_T]

An model of selected items, with a active and current item.

There can only be one active and one current item, but there can be multiple selected items. An "active" item is defined as a single selected item (if multiple items are selected, there is no active item). The "current" item is mostly useful for (e.g.) keyboard actions: even with multiple items selected, you may only have one current item, and keyboard events (like up and down) can modify that current item. It's possible to have a current item without an active item, but an active item will always be the current item.

An item can be the current item and selected at the same time. Qt views will ensure that there is always a current item as keyboard navigation, for example, requires a current item. This pattern mimics current/selected items from Qt: https://doc.qt.io/qt-5/model-view-programming.html#current-item-and-selected-items

Parameters:

• data #

  (iterable, default: () ) –

  Elements to initialize the set with.

• parent #

  (Container, default: None ) –

  The parent container, if any. This is used to provide validation upon mutation in common use cases.

Attributes:

• events (SelectionEvents) –

  SignalGroup that with events related to selection changes. (see SelectionEvents)

• active ((Any, optional)) –

  The active item, if any. An "active" item is defined as a single selected item (if multiple items are selected, there is no active item)

• _current ((Any, optional)) –

  The current item, if any. This is used primarily by GUI views when handling mouse/key events.

Methods:

• clear –

  Clear the selection.

• replace_selection –

  Replace the current selection with new_selection.

• select_only –

  Unselect everything but obj. Add to selection if not currently selected.

• toggle –

  Toggle selection state of obj.

78
79
80
81
82
83

def __init__(self, data: Iterable[_T] = (), parent: Container | None = None):
    self._active: _T | None = None
    self._current_: _T | None = None
    self._parent: Container | None = parent
    super().__init__(iterable=data)
    self._update_active()

active: _T | None

clear(keep_current: bool = False) -> None

116
117
118
119
120
121
122
123
124
125
126

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()

replace_selection(new_selection: Iterable[_T]) -> None

138
139
140
141
142
143
144
145
146

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)

select_only(obj: _T) -> None

132
133
134
135
136

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)

toggle(obj: _T) -> None

128
129
130

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

Bases: SignalGroup

Events available on EventedSet.

Attributes:

• items_changed (added (Tuple[Any, ...], removed: Tuple[Any, ...])) –

  A signal that will emitted whenever an item or items are added or removed. Connected callbacks will be called with callback(added, removed), where added and removed are tuples containing the objects that have been added or removed from the set.

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431

def __init__(self, instance: Any = None) -> None:
    cls = type(self)
    if not hasattr(cls, "_psygnal_signals"):
        raise TypeError(
            "Cannot instantiate `SignalGroup` directly.  Use a subclass instead."
        )

    self._psygnal_instances = {
        name: (
            sig._create_signal_instance(self)
            if name in cls._psygnal_name_conflicts
            else sig.__get__(self, cls)
        )
        for name, sig in cls._psygnal_signals.items()
    }
    self._psygnal_relay = SignalRelay(self._psygnal_instances, instance)