evlib.dataloaders.utils

Shared helpers for dataloader indexing, caching, and optional decoding.

Functions

decode_in_parallel(paths, decoder, *, ...)

Decode files with a bounded thread pool.

find_nearest_index(timestamps, t)

Return the index of the timestamp nearest to t.

freeze_array(arr)

Make an array read only if it exists.

get_flow_coordinate_grid(height, width)

Return cached base coordinate grids for dense flow propagation.

normalize_index(index, sequence_length, ...)

Normalize one Python style sequence index and validate bounds.

normalize_indices(indices, sequence_length, ...)

Normalize many Python style sequence indices and validate bounds.

propagate_flow_step(x_flow, y_flow, ...)

Propagate pixel coordinates through one GT flow field in place.

sample_flow_nearest(x_flow, y_flow, ...)

Sample dense flow at floating coordinates with nearest-neighbor.

sample_flow_nearest_numpy(x_flow, y_flow, ...)

Sample dense flow at floating coordinates with NumPy nearest neighbor.

validate_index_interval(start_index, ...)

Validate a half open [start_index, end_index) interval.

Classes

LazyDecodeCache(max_items)

Small process local LRU cache for decoded file payloads.
evlib.dataloaders.utils.freeze_array(arr)

Make an array read only if it exists.

Parameters:

arr (ndarray[Any, dtype[generic]] | None)

Return type:

None

class evlib.dataloaders.utils.LazyDecodeCache(max_items)

Small process local LRU cache for decoded file payloads.

Parameters:

max_items (int)

get(index)

Return a cached decoded value by index, or None on a miss.

Parameters:

index (int)

Return type:

_DecodedValueT | None

put(index, decoded_value)

Store and return a decoded value.

Parameters:

  • index (int)

  • decoded_value (_DecodedValueT)

Return type:

_DecodedValueT

clear()

Remove all cached values.

Return type:

None

evlib.dataloaders.utils.decode_in_parallel(paths, decoder, *, max_workers)

Decode files with a bounded thread pool.

Parameters:
Return type:

List[Any]

evlib.dataloaders.utils.normalize_index(index, sequence_length, item_name)

Normalize one Python style sequence index and validate bounds.

Parameters:

  • index (int)

  • sequence_length (int)

  • item_name (str)

Return type:

int

evlib.dataloaders.utils.normalize_indices(indices, sequence_length, item_name)

Normalize many Python style sequence indices and validate bounds.

Parameters:
Return type:

ndarray[Any, dtype[int64]]

evlib.dataloaders.utils.validate_index_interval(start_index, end_index, sequence_length, item_name)

Validate a half open [start_index, end_index) interval.

Parameters:

  • start_index (int)

  • end_index (int)

  • sequence_length (int)

  • item_name (str)

Return type:

Tuple[int, int]

evlib.dataloaders.utils.find_nearest_index(timestamps, t)

Return the index of the timestamp nearest to t.

Parameters:
Return type:

int

evlib.dataloaders.utils.get_flow_coordinate_grid(height, width)

Return cached base coordinate grids for dense flow propagation.

Parameters:
Return type:

Tuple[ndarray[Any, dtype[float32]], ndarray[Any, dtype[float32]]]

evlib.dataloaders.utils.sample_flow_nearest_numpy(x_flow, y_flow, x_coords, y_coords)

Sample dense flow at floating coordinates with NumPy nearest neighbor.

Parameters:
Return type:

Tuple[ndarray[Any, dtype[float32]], ndarray[Any, dtype[float32]], ndarray[Any, dtype[bool_]]]

evlib.dataloaders.utils.sample_flow_nearest(x_flow, y_flow, x_coords, y_coords)

Sample dense flow at floating coordinates with nearest-neighbor.

Parameters:
Return type:

Tuple[ndarray[Any, dtype[float32]], ndarray[Any, dtype[float32]], ndarray[Any, dtype[bool_]]]

evlib.dataloaders.utils.propagate_flow_step(x_flow, y_flow, x_coords, y_coords, x_mask, y_mask, scale)

Propagate pixel coordinates through one GT flow field in place.

Parameters:
Return type:

None