leads.data_persistence.core

leads.data_persistence.core#

Module Contents#

Classes#

DataPersistence

Vector

CSV

CSVDataset

Functions#

weighed_sum

weighed_mean

mean_compressor

A compression method that reduces data memory usage by averaging adjacent numbers and merging them. :param sequence: the sequence to compress :param target_size: the expected size :return: the compressed sequence

Data#

T

E

DEFAULT_HEADER

DEFAULT_HEADER_FULL

VISUAL_HEADER_ONLY

VISUAL_HEADER

VISUAL_HEADER_FULL

API#

leads.data_persistence.core.T = '_TypeVar(...)'#
leads.data_persistence.core.weighed_sum(elements: tuple[leads.data_persistence.core.T, ...], indexes: tuple[float, ...], a: int = 0, b: int | None = None) → leads.data_persistence.core.T[source]#
leads.data_persistence.core.weighed_mean(elements: tuple[leads.data_persistence.core.T, ...], indexes: tuple[float, ...], a: int = 0, b: int | None = None) → leads.data_persistence.core.T[source]#
leads.data_persistence.core.mean_compressor(sequence: dict[leads.data_persistence.core.T, float], target_size: int) → dict[leads.data_persistence.core.T, float][source]#

A compression method that reduces data memory usage by averaging adjacent numbers and merging them. :param sequence: the sequence to compress :param target_size: the expected size :return: the compressed sequence

class leads.data_persistence.core.DataPersistence(max_size: int = -1, crop_ratio: int = 2, compressor: leads.types.Compressor[leads.data_persistence.core.T] = mean_compressor)[source]#

Bases: typing.Sequence[leads.data_persistence.core.T], typing.Generic[leads.data_persistence.core.T]

Initialization

Parameters:

  • max_size – the maximum cached size

  • crop_ratio – new size = max size / crop ratio

  • compressor – the compressor interface

__len__() → int[source]#
__getitem__(index: int) → leads.data_persistence.core.T | list[leads.data_persistence.core.T][source]#
__str__() → str[source]#
sum() → leads.data_persistence.core.T[source]#
indexes() → list[float][source]#
weights() → list[float][source]#
to_list() → list[leads.data_persistence.core.T][source]#
append(element: leads.data_persistence.core.T) → None[source]#
leads.data_persistence.core.E = '_TypeVar(...)'#
class leads.data_persistence.core.Vector(*coordinates: leads.data_persistence.core.E)[source]#

Bases: typing.Sequence[leads.data_persistence.core.E], typing.Iterable[leads.data_persistence.core.E], typing.Generic[leads.data_persistence.core.E]

Initialization

__hash__() → int[source]#
__len__() → int[source]#
__iter__() → Iterator[leads.data_persistence.core.E][source]#
__getitem__(item: int | slice) → Self[source]#
__str__() → str[source]#
__eq__(other: Self) → bool[source]#
_check_dimension(other: Self) → None[source]#
__neg__() → Self[source]#
__abs__() → Self[source]#
_operate(other: Self | leads.data_persistence.core.E, operator: Callable[[leads.data_persistence.core.E, leads.data_persistence.core.E], leads.data_persistence.core.E]) → Self[source]#
__add__(other: Self | leads.data_persistence.core.E) → Self[source]#
__sub__(other: Self | leads.data_persistence.core.E) → Self[source]#
__mul__(other: Self | leads.data_persistence.core.E) → Self[source]#
__truediv__(other: Self | leads.data_persistence.core.E) → Self[source]#
__floordiv__(other: Self | leads.data_persistence.core.E) → Self[source]#
distance(other: Self) → float[source]#
magnitude() → float[source]#
_compare(other: Self | leads.data_persistence.core.E, comparer: Callable[[leads.data_persistence.core.E, leads.data_persistence.core.E], bool]) → bool[source]#
__lt__(other: Self | leads.data_persistence.core.E) → bool[source]#
__le__(other: Self | leads.data_persistence.core.E) → bool[source]#
__gt__(other: Self | leads.data_persistence.core.E) → bool[source]#
__ge__(other: Self | leads.data_persistence.core.E) → bool[source]#
class leads.data_persistence.core.CSV(file: str | TextIO, header: tuple[str, ...], *columns: leads.data_persistence.core.DataPersistence | None)[source]#

Bases: object

Initialization

header() → tuple[str, ...][source]#
write_header() → None[source]#
write_frame(*data: Any) → None[source]#
close() → None[source]#
class leads.data_persistence.core.CSVDataset(file: str, chunk_size: int = 100)[source]#

Bases: typing.Iterable[dict[str, typing.Any]]

Initialization

require_loaded() → None[source]#
read_header() → tuple[str, ...][source]#
__iter__() → Generator[dict[str, Any], None, None][source]#
__len__() → int[source]#
load() → None[source]#
save(file: str | TextIO) → None[source]#
close() → None[source]#
leads.data_persistence.core.DEFAULT_HEADER: leads.types.DefaultHeader = ('t', 'voltage', 'speed', 'front_wheel_speed', 'rear_wheel_speed', 'yaw', 'pitch', 'roll', 'forward_...#
leads.data_persistence.core.DEFAULT_HEADER_FULL: leads.types.DefaultHeaderFull = None#
leads.data_persistence.core.VISUAL_HEADER_ONLY: tuple[str, str, str, str, str, str, str, str] = ('front_view_base64', 'front_view_latency', 'left_view_base64', 'left_view_latency', 'right_view_bas...#
leads.data_persistence.core.VISUAL_HEADER: leads.types.VisualHeader = None#
leads.data_persistence.core.VISUAL_HEADER_FULL: leads.types.VisualHeaderFull = None#
Contents