Source#

class deeptrack.sources.base.Source(**kwargs: Sequence[Any])#

Bases: object

A class that represents one or more sources of data.

Source holds one or more named sequences (e.g., lists, arrays) and makes them accessible by index. It returns SourceItem objects that activate registered callbacks (e.g., for dependency tracking) when called.

Each named field is accessible as an attribute (e.g., source.a) and can be passed directly to DeepTrack2 features such as Value. Features can then be evaluated on specific items by indexing the source (e.g., feature(source[i])).

Parameters#

**kwargs: Sequence[Any]: Named data sources, where each key is the name of a source (e.g., “x”, “label”) and each value is an indexable sequence (e.g., list, NumPy array, PyTorch tensor). All sequences must have the same length and support integer indexing.

Attributes#

_dict: dict[str, Sequence[Any]]: Internal mapping of source names to their corresponding data sequences.
_length: int: Number of items in the source. All fields must have the same length.
_current_index: DeepTrackNode: A node that holds the current active index. Used for dynamic access when a source attribute (e.g., source.a) is passed to a feature.
_callbacks: set[Callable[[SourceItem], None]]: A set of callback functions triggered when a SourceItem is called.

Methods#

product(**kwargs: Sequence[Any]) -> Product: Return a new source representing the cartesian product of the current source with the given sequences.
constants(**kwargs: Sequence[Any]) -> Product: Return a new source where the given values are treated as constants.
filter(predicate: Callable[…, bool]) -> Subset: Return a new source containing only the items for which the predicate returns True.
set_index(index) -> Source: Set the active index used when evaluating attributes, like in source.a().

Callback registration. on_activate(callback: Callable[[SourceItem], None]) -> None

Register a callback to be called when any item is activated.

Private and internal methods. __len__() -> int

Return the number of items in the source.

__getitem__(index) -> SourceItem or list[SourceItem]: Retrieve one or more items by index or slice.
_get_item(index: int) -> SourceItem: Retrieve a single SourceItem at a specified index.
_get_slice(slice_obj) -> list[SourceItem]: Retrieve a list of SourceItems corresponding to a slice.
_validate_all_same_length(kwargs) -> None: Validate that all input sequences have the same length.
_wrap(key) -> SourceDeepTrackNode: Wrap a field from the source into a SourceDeepTrackNode.
_wrap_indexable(key) -> SourceDeepTrackNode: Wrap an indexable field as a SourceDeepTrackNode.
_wrap_iterable(key) -> SourceDeepTrackNode: Wrap a non-indexable iterable field as a SourceDeepTrackNode.
__iter__() -> Generator[SourceItem, None, None]: Iterate over all items in the source.
__repr__() -> str:: Return a string representation of the source object.

Examples#

>>> import deeptrack as dt
>>> from deeptrack.sources import Source

Define a source with two fields:

>>> source = Source(
...     a=[1, 2, 3, 4, 5, 6, 7, 8, 9],
...     b=[10, 20, 30, 40, 50, 60, 70, 80, 90],
... )

Create features from the source:

>>> feature_a = dt.Value(source.a)
>>> feature_b = dt.Value(source.b)
>>> sum_feature = feature_a + feature_b

Evaluate features on individual items:

>>> sum_feature(source[0])
11

>>> sum_feature(source[8])
99

Filter items using a predicate:

>>> filtered = source.filter(lambda a, b: a > 5 and b < 80)
>>> list(filtered)
[SourceItem({'a': 6, 'b': 60}, 1 callback(s)),
 SourceItem({'a': 7, 'b': 70}, 1 callback(s))]

Slice the source:

>>> subset = source[3:5]
>>> subset
[SourceItem({'a': 4, 'b': 40}, 1 callback(s)),
 SourceItem({'a': 5, 'b': 50}, 1 callback(s))]

Add a constant field to the source:

>>> augmented = source.constants(label="train")
>>> augmented[0]["label"]
'train'

Take a Cartesian product with a new field:

>>> extended = source.product(c=[100, 200])
>>> len(extended)
18  # 9 original items x 2 values in "c"

>>> extended[0]["c"]
100

>>> extended[17]["c"]
200

Use set_index to manually select the active item:

>>> source.set_index(1)
>>> source.a()
2

>>> source.b()
20

Iterate over items in the source:

>>> for item in source:
...     print(item["a"], item["b"])
10
20
30
40
50
60
70
80
90

Methods Summary

`constants`(**kwargs)	New source where the given values are treated as constants.
`filter`(predicate)	New source containing only items that satisfy a predicate.
`on_activate`(callback)	Register a callback to be triggered when a SourceItem is activated.
`product`(**kwargs)	Cartesian product of the current source with additional fields.
`set_index`(index)	Set the active index of the source for dynamic evaluation.

Methods Documentation

constants(**kwargs: Sequence[Any]) → Product#

New source where the given values are treated as constants.

This method extends the current source with one or more constant fields. Each value is repeated to match the length of the existing source.

Parameters#

**kwargs: Sequence[Any]: Named constant values to add to the source. Each key defines the name of a new field, and each value will be broadcasted as a constant (e.g., scalar, string, etc.).

Returns#

Product: A new source that includes the constant fields in addition to the original fields.

Examples#

>>> from deeptrack.sources import Source

Create a source:

>>> source = Source(a=[1, 2], b=[3, 4])

Add a constant field:

>>> new_source = source.constants(c=5)
>>> new_source
Product(c=[5, 5], a=[1, 2], b=[3, 4])

filter(predicate: Callable[[...], bool]) → Subset#

New source containing only items that satisfy a predicate.

This method filters the source based on a boolean-valued predicate applied to each SourceItem. The result is a Subset containing only the items for which the predicate returns True.

Parameters#

predicate: Callable[…, bool]: A function that takes the fields of a SourceItem as keyword arguments and returns True if the item should be included.

Returns#

Subset: A new source containing only the filtered items.

Examples#

>>> from deeptrack.sources import Source

Create a source:

>>> source = Source(a=[1, 2], b=[3, 4])

Filter to keep only items where a > 1:

>>> new_source = source.filter(lambda a, b: a > 1)
>>> new_source
Subset(a=[2], b=[4])

on_activate(callback: Callable[[SourceItem], None]) → None#

The callback will be executed every time a SourceItem produced by this Source is called (i.e., when item() is invoked). The callback receives the SourceItem as its argument, allowing access or mutation of its contents.

Parameters#

callbackCallable[[SourceItem], None]: A function that takes a SourceItem and performs a side-effect (e.g., logging, modifying metadata, triggering updates). The function must return None.

Examples#

>>> from deeptrack.sources import Source

Define a callback function:

>>> def log_access(item):
...     print(f"CALLBACK - Item accessed: {item}")

Create a source and register the callback:

>>> source = Source(a=[1, 2], b=[10, 20])
>>> source.on_activate(log_access)

>>> item = source[0]
>>> item();
CALLBACK - Item accessed: SourceItem({'a': 1, 'b': 10}, 2 callback(s))

product(**kwargs: Sequence[Any]) → Product#

Cartesian product of the current source with additional fields.

This method returns a new Product source formed by taking the Cartesian product of the current source with the provided sequences. The new source will contain one item for every combination of the original items and the new sequences.

Parameters#

**kwargs: Sequence[Any]: One or more additional sequences to combine with the current source. The keys define the names of the new fields, and the values are indexable sequences (e.g., lists or arrays).

Returns#

Product: A new source representing the Cartesian product of the current source with the additional sequences.

Examples#

>>> from deeptrack.sources import Source

Create an initial source:

>>> source = Source(a=[1, 2], b=[3, 4])

Take the product with a new sequence:

>>> new_source = source.product(c=[5, 6])
>>> new_source
Product(c=[5, 6, 5, 6], a=[1, 1, 2, 2], b=[3, 3, 4, 4])

set_index(index: int) → Source#

Set the active index of the source for dynamic evaluation.

This method updates the internal ._current_index() node, which is used when evaluating attribute-based access such as source.a(). It is typically called automatically when a SourceItem is activated, but can also be called manually to override the index.

Parameters#

index: int: The index to set as the current active index.

Returns#

Source: The source itself, allowing method chaining.

Examples#

>>> from deeptrack.sources import Source

Create a source:

>>> source = Source(
...     a=[1, 2, 3, 4, 5, 6, 7, 8, 9],
...     b=[10, 20, 30, 40, 50, 60, 70, 80, 90],
... )

>>> source.a(), source.b()
(1, 10)

>>> source.set_index(5)
>>> source.a(), source.b()
(6, 60)

>>> source.set_index(-1)
>>> source.a(), source.b()
(9, 90)

>>> source.set_index(1)
>>> source.a(), source.b()
(2, 20)