mdx.mtmc.core.data module

class Loader(config: AppConfig)

Bases: object

Module to load data

Parameters:: config (AppConfig) – configuration for the app

loader = Loader(config)

load_json_data_to_frames(json_data_path: str) → List[Frame]

Loads JSON data from the perception pipeline

Parameters:: json_data_path (str) – path to the JSON data file
Returns:: list of frames
Return type:: List[Frame]

frames = loader.load_json_data_to_frames(json_data_path)

load_protobuf_data_to_frames(protobuf_data_path: str) → List[Frame]

Loads protobuf data from the perception pipeline

Parameters:: protobuf_data_path (str) – path to the protobuf data file
Returns:: list of frames
Return type:: List[Frame]

frames = loader.load_protobuf_data_to_frames(protobuf_data_path)

load_protobuf_message_to_frame(protobuf_frame: Any) → Frame | None

Loads a protobuf frame object to a frame object

Parameters:: frame (Any) – protobuf frame object
Returns:: frame object or None
Return type:: Optional[Frame]

frame = loader.load_protobuf_message_to_frame(protobuf_frame)

load_protobuf_string_to_frame(protobuf_string: str) → Frame | None

Loads a protobuf string to a frame object

Parameters:: protobuf_string (str) – protobuf string
Returns:: frame object or None
Return type:: Optional[Frame]

frame = loader.load_protobuf_string_to_frame(protobuf_string)

class Preprocessor(config: AppConfig)

Bases: object

Module to preprocess frames into behaviors

Parameters:: config (AppConfig) – configuration for the app

preprocessor = Preprocessor(config)

calculate_bbox_area(bbox: Bbox) → float

Calculates bbox area

Parameters:: bbox (Bbox) – bounding box
Returns:: area of the bbox
Return type:: float

calculate_bbox_aspect_ratio(bbox: Bbox) → float

Calculates bbox aspect ratio

Parameters:: bbox (Bbox) – bounding box
Returns:: area of the bbox
Return type:: float

create_behaviors_from_frames(frames: List[Frame]) → List[Behavior]

Creates behaviors from frames

Parameters:: frames (List[Frame]) – list of frames
Returns:: list of behaviors
Return type:: List[Behavior]

behaviors = preprocessor.create_behaviors_from_frames(frames)

create_behaviors_from_protobuf_frames(protobuf_frames: List[Any]) → List[Behavior]

Creates behaviors from protobuf frames

Parameters:: protobuf_frames (List[Any]) – protobuf frames
Returns:: list of behaviors
Return type:: List[Behavior]

behaviors = preprocessor.create_behaviors_from_protobuf_frames(protobuf_frames)

delay_behavior_timestamps(behaviors: List[Behavior]) → List[Behavior]

Delays timestamps of behaviors

Parameters:: List[Behavior] – list of behaviors
Returns:: behaviors after delaying timestamps of behaviors
Return type:: List[Behavior]

delayed_behaviors = preprocessor.delay_behavior_timestamps(behaviors)

filter(behaviors: List[Behavior]) → List[Behavior]

Filters behaviors

Parameters:: behaviors (List[Behavior]) – list of behaviors
Returns:: filtered list of behaviors
Return type:: List[Behavior]

filtered_behaviors = preprocessor.filter(behaviors)

filter_by_bbox(behavior: Behavior) → bool

Filters behavior by bounding boxes

Parameters:: behavior (Behavior) – behavior object
Returns:: decision to filter the behavior
Return type:: bool

filtered_behavior = preprocessor.filter_by_bbox(behavior)

filter_by_behavior_length(behavior: Behavior) → bool

Filters behavior by behavior length in time

Parameters:: behavior (Behavior) – behavior object
Returns:: decision to filter the behavior
Return type:: bool

filtered_behavior = preprocessor.filter_by_behavior_length(behavior)

filter_by_confidence(behavior: Behavior) → bool

Filters behavior by mean detection confidence

Parameters:: behavior (Behavior) – behavior object
Returns:: decision to filter the behavior
Return type:: bool

filtered_behavior = preprocessor.filter_by_confidence(behavior)

group_behaviors_by_places(behaviors: List[Behavior]) → Dict[str, List[Behavior]]

Groups behaviors by places

Parameters:: behaviors (List[Behavior]) – list of behaviors
Returns:: map from places to behaviors
Return type:: Dict[str, List[Behavior]]

map_place_to_behaviors = preprocessor.group_behaviors_by_places(behaviors)

normalize_embeddings(embeddings: List[List[float]]) → List[List[float]]

Normalizes the embeddings of a behavior

Parameters:: embeddings (List[List[float]]) – list of embeddings
Returns:: normalized embeddings
Return type:: List[List[float]]

normalized_embeddings = preprocessor.normalize_embeddings(embeddings)

preprocess(frames: List[Frame]) → List[Behavior]

Preprocesses frames into behaviors and filters outliers

Parameters:: frames (List[Frame]) – list of frames
Returns:: list of behaviors
Return type:: List[Behavior]

behaviors = preprocessor.preprocess(frames)

sample_locations(behavior: Behavior) → Behavior

Sample locations of a long behavior

Parameters:: behavior (Behavior) – behavior object
Returns:: behavior after the locations being sampled
Return type:: Behavior

sampled_behavior = preprocessor.sample_locations(behavior)

set_people_height_estimator(people_height_estimator: StateManager) → None

Sets people height estimator

Parameters:: people_height_estimator (PeopleHeightEstimator) – estimator of people height
Returns:: None

preprocessor.set_people_height_estimator(people_height_estimator)

set_sensor_state_objects(sensor_state_objects: Dict[str, SensorStateObject]) → None

Sets sensor state objects

Parameters:: sensor_state_objects (Dict[str,SensorStateObject]) – map from sensor IDs to sensor state objects
Returns:: None

preprocessor.set_sensor_state_objects(sensor_state_objects)

stitch_behaviors(behaviors: List[Behavior]) → List[Behavior]

Stitches behaviors sharing the same behavior IDs

Parameters:: List[Behavior] – list of behaviors
Returns:: behaviors after stitching
Return type:: List[Behavior]

stitched_behaviors = preprocessor.stitch_behaviors(behaviors)

sum_embeddings(behaviors: List[Behavior]) → List[Behavior]

Sums embeddings of behaviors

Parameters:: behaviors (List[Behavior]) – list of behaviors
Returns:: behaviors with summed embeddings
Return type:: List[Behavior]

behaviors = preprocessor.sum_embeddings(behaviors)

update_people_height(behaviors: List[Behavior]) → None

Updates people’s height

Parameters:: behaviors (List[Behavior]) – list of behaviors
Returns:: None

preprocessor.update_people_height(behaviors)

calculate_bbox_area(bbox: Bbox) → float

Calculates bbox area

Parameters:: bbox (Bbox) – bounding box
Returns:: area of the bbox
Return type:: float

calculate_bbox_aspect_ratio(bbox: Bbox) → float

Calculates bbox aspect ratio

Parameters:: bbox (Bbox) – bounding box
Returns:: area of the bbox
Return type:: float

normalize_vector(vector: List[float]) → numpy.array | None

Normalizes a vector

Parameters:: vector (List[float]) – vector
Returns:: normalized vector or None
Return type:: Optional[List[float]]

perspective_transform(pixel: List[float], homography: List[List[float]]) → List[float] | None

Transforms a 2D pixel to a location on the ground plane

Parameters:

pixel (List[float]) – 2D pixel location
homography (List[List[float]]) – 3x3 homography matrix

Returns:

location on the ground plane

Return type:

Optional[List[float]]