leads.data_persistence.analyzer.inference

Module Contents

Classes

Inference
SpeedInferenceBase
SafeSpeedInference	Infer the speed based on the front wheel speed or the rear wheel speed.
SpeedInferenceByAcceleration	Infer the speed based on the acceleration.
SpeedInferenceByMileage	Infer the speed based on the mileage.
SpeedInferenceByGPSGroundSpeed	Infer the speed based on the GPS ground speed.
SpeedInferenceByGPSPosition	Infer the speed based on the GPS position. This is equivalent to inferring the mileage based on GPS position and then inferring the speed based on the mileage.
ForwardAccelerationInferenceBase
ForwardAccelerationInferenceBySpeed	Infer the forward acceleration based on the speed. Note that this is not always reliable because speed is a scalar, but forward acceleration is not. Accelerating in reverse will still be counted as forward acceleration.
MileageInferenceBase
MileageInferenceBySpeed	Infer the mileage based on the speed.
MileageInferenceByGPSPosition	Infer the mileage based on the speed.
VisualDataRealignmentByLatency	Offset the delay introduced by camera recording and video encoding so that the sensor data and the picture of the same frame match.
InferredDataset

API

class leads.data_persistence.analyzer.inference.Inference(required_depth: tuple[int, int] = (0, 0), required_header: tuple[str, ...] = DEFAULT_HEADER)

Bases: object

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

depth() → tuple[int, int]

Returns:

(-depth backward, depth forward)

header() → tuple[str, ...]

abstractmethod complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

Infer, based on the data flow, to complete the missing columns. :param rows: the data flow with the length of depth backward + depth forward :param backward: True: globally reversed; False: regular :return:

class leads.data_persistence.analyzer.inference.SpeedInferenceBase(required_depth: tuple[int, int] = (0, 0), required_header: tuple[str, ...] = DEFAULT_HEADER)

Bases: leads.data_persistence.analyzer.inference.Inference

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

static skip(row: dict[str, Any]) → bool

class leads.data_persistence.analyzer.inference.SafeSpeedInference

Bases: leads.data_persistence.analyzer.inference.SpeedInferenceBase

Infer the speed based on the front wheel speed or the rear wheel speed.

v = min(fws, rws)

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.SpeedInferenceByAcceleration

Bases: leads.data_persistence.analyzer.inference.SpeedInferenceBase

Infer the speed based on the acceleration.

v = ∫a(t)dt

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.SpeedInferenceByMileage

Bases: leads.data_persistence.analyzer.inference.SpeedInferenceBase

Infer the speed based on the mileage.

v = ds/dt

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.SpeedInferenceByGPSGroundSpeed

Bases: leads.data_persistence.analyzer.inference.SpeedInferenceBase

Infer the speed based on the GPS ground speed.

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.SpeedInferenceByGPSPosition

Bases: leads.data_persistence.analyzer.inference.SpeedInferenceBase

Infer the speed based on the GPS position. This is equivalent to inferring the mileage based on GPS position and then inferring the speed based on the mileage.

v = ds/dt

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.ForwardAccelerationInferenceBase(required_depth: tuple[int, int] = (0, 0), required_header: tuple[str, ...] = DEFAULT_HEADER)

Bases: leads.data_persistence.analyzer.inference.Inference

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

static skip(row: dict[str, Any]) → bool

class leads.data_persistence.analyzer.inference.ForwardAccelerationInferenceBySpeed

Bases: leads.data_persistence.analyzer.inference.ForwardAccelerationInferenceBase

Infer the forward acceleration based on the speed. Note that this is not always reliable because speed is a scalar, but forward acceleration is not. Accelerating in reverse will still be counted as forward acceleration.

a = dv/dt

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.MileageInferenceBase(required_depth: tuple[int, int] = (0, 0), required_header: tuple[str, ...] = DEFAULT_HEADER)

Bases: leads.data_persistence.analyzer.inference.Inference

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

static skip(row: dict[str, Any]) → bool

class leads.data_persistence.analyzer.inference.MileageInferenceBySpeed

Bases: leads.data_persistence.analyzer.inference.MileageInferenceBase

Infer the mileage based on the speed.

s = ∫v(t)dt

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.MileageInferenceByGPSPosition

Bases: leads.data_persistence.analyzer.inference.MileageInferenceBase

Infer the mileage based on the speed.

s = s_0 + Δs

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.VisualDataRealignmentByLatency(*channels: Literal[front, left, right, rear])

Bases: leads.data_persistence.analyzer.inference.Inference

Offset the delay introduced by camera recording and video encoding so that the sensor data and the picture of the same frame match.

Initialization

Declare the scale of data this inference requires. :param required_depth: (-depth backward, depth forward) :param required_header: the necessary header that the dataset must contain for this inference to work

complete(*rows: dict[str, Any], backward: bool = False) → dict[str, Any] | None

class leads.data_persistence.analyzer.inference.InferredDataset(file: str, chunk_size: int = 100)

Bases: leads.data_persistence.core.CSVDataset

Initialization

__len__() → int

static merge(raw: dict[str, Any], inferred: dict[str, Any]) → None

Merge the inferred data to the raw data. Overwrite if conflicts. It directly alters the raw data object. :param raw: the raw data :param inferred: the difference data

_complete(inferences: tuple[leads.data_persistence.analyzer.inference.Inference, ...], enhanced: bool, backward: bool) → int

load() → None

assume_initial_zeros() → None

complete(*inferences: leads.data_persistence.analyzer.inference.Inference, enhanced: bool = False, assume_initial_zeros: bool = False) → int

Infer the missing values in the dataset. :param inferences: the inferences to apply :param enhanced: True: use inferred data to infer other data; False: use only raw data to infer other data :param assume_initial_zeros: True: reasonably set any missing data in the first row to zero; False: no change :return: the number of affected rows

__iter__() → Generator[dict[str, Any], None, None]

close() → None