Components & Customization

To help understand more and potentially customize the microservice, here’s a deep dive of the components’ implementation and the data flow between them:

Library API

The microservice extensively uses API from the MTMC library. The microservice pipeline and the API usage are discussed in-depth in the subsequent sections.

When you build a new pipeline using this library (by modifying ours or creating one from scratch), the library dependencies will be fetched from our artifactory. For more details, refer to the README.md in the metropolis-apps-standalone-deployment/modules/multi-camera-tracking/ directory.

Calibration

The calibration module is used to load and compute information of sensors. The calibrate() function takes a JSON file of Sensor data (scale factor, place, matched image and global coordinates, etc.,) as input. The homography that projects from image coordinates to global coordinates is computed. The output is a dictionary of sensor state objects that are used to compute the human objects’ locations, i.e., trajectories.

calibrate(self, calibration_path: str) → Dict[str, SensorStateObject]

Calibrates sensors

Parameters: calibration_path (str) – path to the calibration file in JSON format
Returns: map from sensor IDs to sensor state objects
Return type: Dict[str,SensorStateObject]

from mdx.mtmc.core.calibration import Calibrator

calibrator = Calibrator(config)
sensor_state_objects = calibrator.calibrate(calibration_path)

Data Loading

The data loading module is used to load raw data in JSON format or protobuf bytes.

from mdx.mtmc.core.data import Loader
loader = Loader(config)

The load_json_data_to_frames() function takes the path to a JSON file as input and loads the data into frame objects.

load_json_data_to_frames(self, 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)

The load_protobuf_string_to_frame() function takes a protobuf frame object as input and creates a frame object.

load_protobuf_string_to_frame(self, protobuf_string: str) → Optional[Frame]
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)

The load_protobuf_message_to_frame() function takes the protobuf messages (bytes) as input creates a frame object.

load_protobuf_message_to_frame(self, protobuf_frame: Any) → Optional[Frame]
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)

The load_protobuf_data_to_frames() function takes the path to a file of protobuf messages (bytes) as input and loads the data into frame objects.

load_protobuf_data_to_frames(self, 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)

Data Preprocessing

The data preprocessing module is used to process Frame objects into behavior objects and filter outliers. Each behavior contains the object ID, locations, bounding boxes, embeddings, etc., of a single-camera tracklet. The preprocess() function first converts the loaded raw data into behavior objects. To save memory, the locations are first sampled to keep the number under a given threshold. Then the locations and embeddings are filtered. We also have behavior-level filters. Finally, the summed and normalized embedding is calculated for each behavior for clustering.

preprocess(self, 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]

from mdx.mtmc.core.data import Preprocessor

preprocessor = Preprocessor(config)
behaviors = preprocessor.preprocess(frames)

Clustering

The clustering module is used to cluster behavior objects into MTMC objects based on appearance features (embeddings from the re-identification model) and spatio-temporal information. Each MTMC object contains the global ID and matched behaviors of a cluster.

from mdx.mtmc.core.clustering import Clusterer
clusterer = Clusterer(config)

The cluster() function first cluster the behaviors based on hierarchical clustering (agglomerative clustering or HDBSCAN). The next step is to re-assign each group of co-existing behaviors, i.e., behaviors that exist at the same time in a sensor that should belong to different clusters, based on the Hungarian algorithm using a combination of appearance distance and spatio-temporal distance. This re-assignment process can be repeated through multiple iterations to further refine the assignment of clusters. Finally, we provide an optional step to suppress the overlapping behaviors, i.e., co-existing behaviors assigned to the same cluster, in the output clusters based on linear programming.

cluster(self, behaviors: List[Behavior]) → Tuple[List[MTMCObject], List[str], np.array]
Clusters behaviors to get MTMC objects

Parameters

behaviors (List[Behavior]) – list of behaviors

Returns

list of MTMC objects, list of behavior keys, and embedding array

Return type

Tuple[List[MTMCObject],List[str],np.array]
mtmc_objects, behavior_keys, embedding_array = clusterer.cluster(behaviors)

The match_online() function matches behaviors with existing clusters (MTMC plus objects in the state) based on the Hungarian algorithm.

match_online(self, behaviors: List[Behavior], mtmc_state_objects_plus: Dict[str, MTMCStateObjectPlus]) → Tuple[List[MTMCObject], Dict[str, List[List[float]]]]
Matches behaviors in an online mode

Parameters

behaviors (List[Behavior]) – list of behaviors

mtmc_state_objects_plus (Dict[str,MTMCStateObjectPlus]) – map from global IDs to MTMC state objects plus

Returns

list of MTMC objects and map from global IDs to mean embeddings

Return type

Tuple[List[MTMCObject],Dict[str,List[List[float]]]]
mtmc_objects, map_global_id_to_mean_embedding = clusterer.match_online(behaviors, mtmc_state_objects_plus)

The stitch_mtmc_objects_with_state() function stitches re-initialized MTMC objects with existing MTMC state objects plus to keep the global IDs consistent.

match_online(self, behaviors: List[Behavior], mtmc_state_objects_plus: Dict[str, MTMCStateObjectPlus]) → Tuple[List[MTMCObject], Dict[str, List[List[float]]]]
Stitches MTMC objects with existing MTMC state objects plus

Parameters

mtmc_objects (List[MTMCObject]) – list of MTMC objects

mtmc_state_objects_plus (Dict[str,MTMCStateObjectPlus]) – map from global IDs to MTMC state objects plus

Returns

updated list of MTMC objects

Return type

List[MTMCObject]
mtmc_objects = clusterer.stitch_mtmc_objects_with_state(mtmc_objects, mtmc_state_objects_plus)

Evaluation

The module for evaluation computes the HOTA metrics (HOTA, Detection Accuracy, Association Accuracy, Localization Accuracy, etc.), CLEAR metrics (MOT Accuracy, MOT Precision, etc.) and identity metrics (ID F1, ID Recall, ID Precision, etc.), if the ground truth is available. The evaluate() function can resolve duplicate assignments of behaviors before computing the accuracy scores. We also provide options for (1) measuring accuracy based on full-body bounding boxes, (2) using metrics based on 3D distance on the ground plane, and (3) plotting diagrams for evaluation. Please refer to the HOTA paper for details of the evaluation metrics.

evaluate(self, frames: List[Frame], mtmc_objects: List[MTMCObject], sensor_state_objects: Dict[str, SensorStateObject]) → None

Evaluates MTMC results

Parameters

frames (List[Frame]) – list of frames
mtmc_objects (List[MTMCObject]) – list of MTMC objects
sensor_state_objects (Dict[str,SensorStateObject]) – map from sensor IDs to sensor state objects

Return type

None

from mdx.mtmc.core.evaluation import Evaluator

evaluator = Evaluator(config)
evaluator.evaluate(frames, mtmc_objects, sensor_state_objects)

Kafka Messaging

This module is used to consume and produce messages using Spark Structured Streaming.

from mdx.mtmc.stream.kafka_message_broker import KafkaMessageBroker
kafka_message_broker = KafkaMessageBroker(config)

The set_sensor_state_objects() function sets the sensor state objects for calibration.

set_sensor_state_objects(self, 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
kafka_message_broker.set_sensor_state_objects(sensor_state_objects)

The get_consumed_raw_messages_and_count() function consumes messages and count from the mdx-raw topic.

get_consumed_raw_messages_and_count(self, consumer: KafkaConsumer) → Tuple[List[Any], int]
Consumes messages and count from the mdx-raw topic

Parameters

consumer (KafkaConsumer) – Kafka consumer

Return type

Tuple[List[Any],int]
messages, message_count = kafka_message_broker.get_consumed_raw_messages_and_count(consumer)

The get_consumed_raw_messages() function consumes partitioned messages from the mdx-raw topic.

get_consumed_raw_messages(self, consumer: KafkaConsumer) → Dict[TopicPartition, List[KafkaConsumerRecord]]
Consumes partitioned messages from the mdx-raw topic

Parameters

consumer (KafkaConsumer) – Kafka consumer

Return type

Dict[TopicPartition,List[KafkaConsumerRecord]]
partitioned_messages = kafka_message_broker.get_consumed_raw_messages(consumer)

The get_consumed_notification_messages_and_count() function consumes messages and count from the mdx-notification topic.

get_consumed_notification_messages_and_count(self, consumer: KafkaConsumer) → Tuple[List[KafkaConsumerRecord], int]
Consumes messages and count from the mdx-notification topic

Parameters

consumer (KafkaConsumer) – Kafka consumer

Returns

Kafka consumer records and the count of messages

Return type

Tuple[List[KafkaConsumerRecord],int]
messages, message_count = kafka_message_broker.get_consumed_notification_messages_and_count(consumer)

The produce_mtmc_messages() function produces messages containing MTMC objects from clustering results and sends them to the mdx-mtmc topic.

produce_mtmc_messages(self, producer: KafkaProducer, mtmc_objects: List[MTMCObject]) → None
Sends messages containing MTMC objects to mdx-mtmc

Parameters

producer (KafkaProducer) – Kafka producer

mtmc_objects (List[MTMCObject]) – list of MTMC objects

Returns

None
kafka_message_broker.produce_mtmc_messages(producer, mtmc_objects)

The produce_mtmc_plus_messages() function produces messages containing MTMC plus objects from online tracking results and sends them to the mdx-rtls topic.

produce_mtmc_plus_messages(self, producer: KafkaProducer, mtmc_objects_plus: MTMCObjectsPlus) → None
Sends messages containing MTMC objects plus locations to mdx-rtls

Parameters

producer (KafkaProducer) – Kafka producer

mtmc_objects_plus (MTMCObjectsPlus) – MTMC objects plus locations

Returns

None
kafka_message_broker.produce_mtmc_plus_messages(producer, mtmc_objects_plus)

The produce_calibration_request() function produces a message requesting calibration and sends it to mdx-notification.

produce_calibration_request(self, producer: KafkaProducer) → None
Sends a message requesting calibration to mdx-notification

Parameters

producer (KafkaProducer) – Kafka producer

Returns

None
kafka_message_broker.produce_calibration_request(producer)

The produce_config_request() function produces a message requesting app configuration and sends it to mdx-notification.

produce_config_request(self, producer: KafkaProducer) → None
Sends a message requesting app config

Parameters

producer (KafkaProducer) – Kafka producer

Returns

None
kafka_message_broker.produce_config_request(producer)

The produce_init_config() function produces a message containing initial app configuration and sends it to mdx-notification.

produce_init_config(self, producer: KafkaProducer) → None
Sends a message containing init app config

Parameters

producer (KafkaProducer) – Kafka producer

Returns

None
kafka_message_broker.produce_init_config(producer)

Data Transformation

This module is for converting Kafka messages into objects used in MTMC microservice.

from mdx.mtmc.stream.transform import DataTransformer
data_transformer = DataTransformer()

The transform_raw_messages_to_protobuf_frames() function transforms messages from the mdx-raw topic to create protobuf frame objects.

transform_raw_messages_to_protobuf_frames(self, messages: List[KafkaConsumerRecord]) → List[Any]
Transforms raw messages to protobuf frame objects

Parameters

messages (List[KafkaConsumerRecord]) – list of Kafka consumer records

Returns

list of protobuf frame objects

Return type

List[Any]
protobuf_frames = data_transformer.transform_raw_messages_to_protobuf_frames(messages)

The transform_raw_messages_to_behaviors() function transforms messages from the mdx-raw topic to create behaviors.

transform_raw_messages_to_behaviors(self, messages: List[bytes], data_preprocessor: Preprocessor) → List[Behavior]
Transforms raw messages to behavior objects

Parameters

messages (List[bytes]) – list of byte messages

data_preprocessor (Preprocessor) – data preprocessor

Returns

list of behaviors

Return type

List[Behavior]
protobuf_frames = data_transformer.transform_raw_messages_to_protobuf_frames(messages)

The transform_notification_messages() function transforms Kafka consumer records from the mdx-notification topic into notification objects.

transform_notification_messages(self, messages: List[KafkaConsumerRecord], calibrator: Calibrator) → Dict[str, Notification]
Transforms Kafka consumer records to notifications

Parameters

messages (List) – list of Kafka consumer records

calibrator (Calibrator) – calibrator used to load sensors

Returns

map from type to notifications

Return type

Dict[str,Notification]
map_type_to_notifications = data_transformer.transform_notification_messages(messages, calibrator)

Multi-Processing

This module is for multi-processing to process raw messages in parallel. The consume_raw_messages_and_add_to_behavior_list() function consumes raw messages from the mdx-raw topic and adds to a behavior list.

consume_raw_messages_and_add_to_behavior_list(self, idx_process: int, behavior_shared_list: List[Behavior]) → None

Consumes raw messages and adds to a behavior list

Parameters

idx_process (int) – index of process
behavior_shared_list (List[Behavior]) – shared list of behaviors

Returns

None

from mdx.mtmc.stream.multiprocessor import MultiProcessor

multi_processor = MultiProcessor(config, data_transformer, data_preprocessor, kafka_message_broker)
multi_processor.consume_raw_messages_and_add_to_behavior_list(idx_process, behavior_shared_list)

Managing Configuration

This module is for managing configuration. The process_notifications_and_update_config() function processes notifications and updates configuration.

process_notifications_and_update_config(self, config_notifications: List[Notification], producer: KafkaProducer)

Processes notifications and updates configuration

Parameters

config_notifications (List[Notification]) – notifications of configuration
producer (KafkaProducer) – Kafka producer

Returns

None

from mdx.mtmc.stream.config_manager import MultiProcessor

config_manager = ConfigManager(config, kafka_message_broker)
config_manager.process_notifications_and_update_config(config_notifications, producer)

Behavior State Management

This module is for managing the state of behaviors.

from mdx.mtmc.stream.state.behavior import StateManager
behavior_state = StateManager(config)

The update_state() function updates behaviors in state and ensures that the resulted behaviors from micro batches are consistent with the expected batch processing results.

update_state(self, behaviors: List[Behavior], preprocessor: Preprocessor) → None
Updates behavior state

Parameters

behaviors (List[Behavior]) – list of behaviors

preprocessor (Preprocessor) – preprocessor used to sum embeddings of a behavior

Returns

None
behavior_state.update_state(behaviors, preprocessor)

The update_places_and_locations() function is used to update places and locations of the behaviors in state with updated sensor information from calibration.

update_places_and_locations(self, sensor_state_objects: Dict[str, SensorStateObject], updated_sensor_ids: Set[str]) → None
Updates places and locations of behavior state objects

Parameters

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

updated_sensor_ids (Set[str]) – updated sensor IDs

Returns

None
behavior_state.update_places_and_locations(sensor_state_objects, updated_sensor_ids)

The get_behavior() function gets the Behavior object given a behavior ID and returns None if the behavior ID cannot be found.

get_behavior(self, behavior_id: str) → Optional[Behavior]
Gets a behavior given the behavior ID

Parameters

behavior_id (str) – behavior ID

Returns

behavior

Return type

Optional[Behavior]
behavior = behavior_state.get_behavior(behavior_id)

The get_behaviors() function gets a list of Behavior objects given multiple behavior IDs.

get_behaviors(self, behavior_ids: List[str]) → List[Optional[Behavior]]
Gets a list of behaviors given the behavior IDs

Parameters

behavior_ids (List[str]) – list of behavior IDs

Returns

list of behaviors

Return type

List[Optional[Behavior]]
behaviors = behavior_state.get_behaviors(behavior_ids)

The get_behaviors_in_state() function gets a list of all the Behavior objects in state.

get_behaviors_in_state(self) → List[Behavior]
Gets all the behaviors in state

Returns

list of behaviors

Return type

List[Behavior]
behaviors = behavior_state.get_behaviors_in_state()

The get_behavior_ids_in_state() function gets a set of all the behavior IDs in state.

get_behavior_ids_in_state(self) → Set[str]
Gets all the behavior IDs in state

Returns

set of behavior IDs

Return type

Set[str]
behavior_ids = behavior_state.get_behavior_ids_in_state()

The get_behavior_keys_in_state() function gets a set of all the behavior keys (sensor-object-timestamp IDs) in state.

get_behavior_keys_in_state(self) → Set[str]
Gets all the behavior keys (sensor-object-timestamp IDs) in state

Returns

set of behavior keys (sensor-object-timestamp IDs)

Return type

Set[str]
behavior_keys = behavior_state.get_behavior_keys_in_state()

The delete_older_state() function deletes older behaviors that are beyond the configuration of retention time limit.

delete_older_state(self) → Set[str]
Deletes behaviors in state that have not been updated for a configurable time interval

Returns

deleted behavior keys

Returns

Set[str]
deleted_behavior_keys = behavior_state.delete_older_state()

MTMC State Management

This module is for managing the state of MTMC objects.

from mdx.mtmc.stream.state.mtmc import StateManager
mtmc_state = StateManager()

The get_global_ids_in_state() function gets a set of all the global IDs in state.

get_global_ids_in_state(self) → Set[str]
Gets all the global IDs in state

Returns

set of global IDs

Return type

Set[str]
global_ids = mtmc_state.get_global_ids_in_state()

The get_mtmc_object() function gets the MTMC object given a global ID and returns None if the global ID cannot be found.

get_mtmc_object(self, global_id: str) → Optional[MTMCObject]
Gets the MTMC object given a global ID

Parameters

global_id (str) – global ID

Returns

MTMC object

Return type

Optional[MTMCObject]
mtmc_object = mtmc_state.get_mtmc_object(global_id)

The get_mtmc_objects() function gets a list of MTMC objects given multiple global IDs.

get_mtmc_objects(self, global_ids: List[str]) → List[Optional[MTMCObject]]
Gets a list of MTMC objects given a list of global IDs

Parameters

global_ids (List[str]) – global IDs

Returns

list of MTMC objects

Return type

List[Optional[MTMCObject]]
mtmc_objects = mtmc_state.get_mtmc_objects(global_ids)

The get_mtmc_objects_in_state() function gets a list of all the MTMC objects in state.

get_mtmc_objects_in_state(self) → List[MTMCObject]
Gets all the MTMC objects in state

Returns

list of MTMC objects

Return type

List[MTMCObject]
mtmc_objects = mtmc_state.get_mtmc_objects_in_state()

The get_map_global_id_to_mtmc_object() function gets a dictionary mapping from all the global IDs to the corresponding MTMC objects in state.

get_map_global_id_to_mtmc_object(self) → Dict[str, MTMCObject]
Gets a map from all the global IDs to MTMC objects in state

Returns

map from global IDs to MTMC objects

Return type

Dict[str,MTMCObject]
global_id_mtmc_object_dict = mtmc_state.get_map_global_id_to_mtmc_object()

The get_updated_mtmc_objects() function updates MTMC objects in state and returns the updated MTMC objects.

get_updated_mtmc_objects(self, mtmc_objects: List[MTMCObject], batch_id: int) → List[MTMCObject]
Gets updated MTMC objects from state

Parameters

mtmc_objects (List[str]) – list of MTMC objects

batch_id (int) – batch ID

Returns

updated MTMC objects

Return type

List[MTMCObject]
updated_mtmc_objects = mtmc_state.get_updated_mtmc_objects(mtmc_objects, batch_id)

The delete_older_state() function deletes older MTMC objects that are beyond the configuration of retention time limit.

delete_older_state(self, keys_in_behavior_state: Set[str]) → None
Deletes global IDs in state when none of the matched behaviors are present in the behavior state

Parameters

keys_in_behavior_state (Set[str]) – keys (ID-timestamp pairs) present in the behavior state

Returns

None
mtmc_state.delete_older_state(keys_in_behavior_state)

MTMC Plus State Management

This module is for managing the state of MTMC plus objects.

from mdx.mtmc.stream.state.mtmc_plus import StateManager
mtmc_plus_state = StateManager(config)

The get_num_mtmc_state_objects_plus() function gets the number of MTMC plus objects in the state.

get_num_mtmc_state_objects_plus(self) → int
Gets the number of MTMC state objects plus

Returns

number of MTMC state objects plus

Return type

int
num_mtmc_state_objects_plus = mtmc_plus_state.get_num_mtmc_state_objects_plus()

The get_mtmc_state_objects_plus() function gets a dictionary of MTMC plus objects in the state.

get_mtmc_state_objects_plus(self) → Dict[str, MTMCStateObjectPlus]
Gets MTMC state objects plus

Returns

MTMC state objects plus

Return type

Dict[str,MTMCStateObjectPlus]
mtmc_state_objects_plus = mtmc_plus_state.get_mtmc_state_objects_plus()

The set_sensor_state_objects() function sets sensor state objects.

set_sensor_state_objects(self, 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
mtmc_plus_state.set_sensor_state_objects(sensor_state_objects)

The check_init_state() function checks whether to (re-)initialize the state.

check_init_state(self, behaviors: List[Behavior]) → bool
Checks whether to (re-)initialize the state

Parameters

mtmc_objects (List[Behavior]) – list of behaviors

Returns

flag indicating that the state needs to be (re-)initialized

Return type

bool
flag_init_state = mtmc_plus_state.check_init_state(behaviors)

The check_init_buffer_ready() function checks whether the initialization buffer for MTMC plus state is ready.

check_init_buffer_ready(self, behaviors: List[Behavior]) → bool
Check whether the buffer of live behaviors for initialization is sufficiently long

Parameters

mtmc_objects (List[Behavior]) – list of MTMC objects

Returns

flag indicating that the buffer of live behaviors for initialization is sufficiently long

Return type

bool
flag_init_buffer_ready = mtmc_plus_state.check_init_buffer_len(behaviors)

The compute_mean_embeddings() function computes mean embeddings of the MTMC plus objects in the state.

compute_mean_embeddings(self, mtmc_objects: List[MTMCObject], behavior_keys: List[str], embedding_array: np.array) → Dict[str, List[List[float]]]
Computes mean embeddings

Parameters

mtmc_objects (List[MTMCObject]) – list of MTMC objects

behavior_keys (List[str]) – list of behvaior keys

embedding_array (np.array) – embedding array

Returns

map from global IDs to mean embeddings

Return type

Dict[str,List[List[float]]]
map_global_id_to_mean_embedding = mtmc_plus_state.compute_mean_embeddings(mtmc_objects, behavior_keys, embedding_array)

The update_mean_embeddings() function updates mean embeddings of the MTMC plus objects in the state.

update_mean_embeddings(self, map_global_id_to_mean_embedding: Dict[str, List[List[float]]]) → Dict[str, List[List[float]]]
Updates mean embeddings

Parameters

map_global_id_to_mean_embedding (Dict[str,List[List[float]]]) – map from global IDs to mean embeddings

Returns

updated map from global IDs to mean embeddings

Return type

Dict[str,List[List[float]]]
updated_map_global_id_to_mean_embedding = mtmc_plus_state.update_mean_embeddings(map_global_id_to_mean_embedding)

The get_mtmc_objects_plus() function gets MTMC plus objects from the state.

get_mtmc_objects_plus(self, mtmc_objects: List[MTMCObject], behaviors: List[Behavior]) → MTMCObjectsPlus
Gets MTMC objects plus locations

Parameters

mtmc_objects (List[MTMCObject]) – list of MTMC objects

behaviors (List[Behavior]) – list of behvaiors

Returns

MTMC objects plus locations

Return type

MTMCObjectsPlus
mtmc_objects_plus = mtmc_plus_state.get_mtmc_objects_plus(mtmc_objects, behaviors)

The update_mtmc_objects_plus() function updates MTMC plus objects using the state.

update_mtmc_objects_plus(self, mtmc_objects_plus: MTMCObjectsPlus, map_global_id_to_mean_embedding: Dict[str, List[List[float]]]) → MTMCObjectsPlus
Update MTMC objects plus with smoothed locations

Parameters

mtmc_objects_plus (MTMCObjectsPlus) – MTMC objects plus locations

map_global_id_to_mean_embedding (Dict[str,List[List[float]]]) – map from global IDs to mean embeddings

Returns

updated MTMC objects plus locations

Return type

MTMCObjectsPlus
updated_mtmc_objects_plus = mtmc_plus_state.get_updated_mtmc_objects_plus(mtmc_objects_plus, map_global_id_to_mean_embedding)

The delete_older_state() function deletes MTMC plus objects in the state.

delete_older_state(self, deleted_behavior_keys: Set[str]) → None
Deletes global IDs in state when none of the matched behaviors are present in the behavior state

Parameters

deleted_behavior_keys (Set[str]) – deleted behavior keys

Returns

None
mtmc_plus_state.delete_older_state(deleted_behavior_keys)

The clean_unused_state() function cleans unused state after stitching global IDs.

clean_unused_state(self, mtmc_objects: List[MTMCObject]) → None
Cleans unused state after stitching global IDs

Parameters

mtmc_objects (List[MTMCObject]) – list of MTMC objects

Returns

None
mtmc_plus_state.clean_unused_state(deleted_behavior_keys)

Sensor State Management

This module is for managing the state of sensors.

from mdx.mtmc.stream.state.sensor import StateManager
sensor_state = StateManager()

The init_state() function initializes state by loading from a local calibration file.

init_state(self, calibrator: Calibrator, calibration_path: str) → None
Initializes state with existing calibration info

Parameters

calibrator (Calibrator) – calibrator used to create sensor state objects

calibration_path (str) – path to the calibration file in JSON format

Returns

None
sensor_state.init_state(calibrator, calibration_path)

The update_state() function updates Sensor objects in state based on a list of Notification objects.

update_state(self, notifications: List[Notification], calibrator: Calibrator) → Set[str]
Updates sensor state

Parameters

notifications (List[Notification]) – list of notifications

calibrator (Calibrator) – calibrator used to convert sensors to sensor state objects

Returns

updated sensor IDs

Return type

Set[str]
updated_sensor_ids = sensor_state.update_state(notifications, calibrator)

The get_sensor_state_objects() function gets a dictionary mapping from all the sensor IDs to the corresponding Sensor objects in state.

get_sensor_state_objects(self) → Dict[str, SensorStateObject]
Gets a map from all the sensor IDs to sensor state objects in state

Returns

map from sensor IDs to sensor state objects

Return type

Dict[str,SensorStateObject]
sensor_state_objects = sensor_state.get_sensor_state_objects()

Estimating People’s Height

This module is for estimating people’s height.

from mdx.mtmc.stream.people_height import StateManager
people_height_state = StateManager(config)

The set_sensor_state_objects() function sets sensor state objects.

set_sensor_state_objects(self, 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
people_height_state.set_sensor_state_objects(sensor_state_objects)

The estimate_people_height() function estimates people’s height from frames.

estimate_people_height(self, frames: List[Frame]) → None
Estimates people’s height from frames

Parameters

frames (List[Frame]) – list of frames

Returns

None
people_height_state.estimate_people_height(frames)

The update_people_height() function updates people’s height in stream processing.

update_people_height(self, behaviors: List[Behavior]) → None
Updates people’s height

Parameters

behaviors (List[Behavior]) – list of behaviors

Returns

None
people_height_state.update_people_height(behaviors)

The rectify_bbox() function computes estimated foot location in pixel and visibility.

rectify_bbox(self, foot_pixel: List[float], bbox: Bbox, sensor_id: str) → Tuple[List[float], float]
Computes expected foot location

Parameters

foot_pixel (List[float]) – foot pixel location

bbox (Bbox) – bounding box

sensor_id (str) – sensor ID

Returns

people’s height

Return type

float
foot_pixel, visibility = people_height_state.rectify_bbox(foot_pixel, bbox, sensor_id)

Visualization

We provide modules for visualizing frames, behaviors, MTMC objects, and ground truth. The data to be visualized can be read from JSON files following the defined schema, which can be generated in batch processing by setting the configuration enableDebug as True. The videos from which the raw data are computed also need to be provided as input.

Visualization of Frames

This module is used to visualize Frame objects. The plot() function plots detected bounding boxes with object IDs frame by frame. The ground truth can also be visualized using this module.

plot(self, sensor_id: str, output_video_dir: str, frames: Dict[str, Dict[str, Dict[str, Object]]], behaviors: Optional[Dict[str, Dict[str, Behavior]]] = None) → None
Visualizes frames

Parameters

sensor_id (str) – sensor ID

output_video_dir (str) – output directory for videos

frames (Dict[str,Dict[str,Dict[str,Object]]]) – frames extracted from raw data

behaviors (Optional[Dict[str,Dict[str,Behavior]]]) – behaviors, i.e., tracklets

Returns

None
from mdx.mtmc.viz.frames import VizFrames

visualizer = VizFrames(config)
visualizer.plot(sensor_id, output_video_dir, frames, behaviors)

Visualization of Behaviors

This module is used to visualize Behavior objects. The plot() function plots each behavior, i.e., tracklet, in the corresponding frame images.

plot(self, sensor_id: str, output_video_dir: str, frames: Dict[str, Dict[str, Dict[str, Object]]], behaviors: Dict[str, Dict[str, Behavior]]) → None
Visualizes behaviors

Parameters

sensor_id (str) – sensor ID

output_video_dir (str) – output directory for videos

frames (Dict[str,Dict[str,Dict[str,Object]]]) – frames extracted from raw data

behaviors (Dict[str,Dict[str,Behavior]]) – behaviors, i.e., tracklets

Returns

None
from mdx.mtmc.viz.behaviors import VizBehaviors

visualizer = VizBehaviors(config)
visualizer.plot(sensor_id, output_video_dir, frames, behaviors)

Visualization of MTMC Objects in Grid

This module is used to visualize MTMC objects in grid mode. The plot() function plots MTMC objects in a grid of synchronized videos.

plot(self, global_id: str, output_video_dir: str, frames: Dict[str, Dict[str, Dict[str, Object]]], behaviors: Dict[str, Dict[str, Behavior]], mtmc_objects: Dict[str, Dict[str, Behavior]]) → None
Visualizes MTMC objects in grid

Parameters

global_id (str) – global ID of an MTMC object

output_video_dir (str) – output directory for videos

frames (Dict[str,Dict[str,Dict[str,Object]]]) – frames extracted from raw data

behaviors (Dict[str,Dict[str,Behavior]]) – behaviors, i.e., tracklets

mtmc_objects (Dict[str,Dict[str,Behavior]]) – MTMC objects, i.e., matches

Returns

None
from mdx.mtmc.viz.mtmc_objects_in_grid import VizMTMCObjectsInGrid

visualizer = VizMTMCObjectsInGrid(config)
visualizer.plot(global_id, output_video_dir, frames, behaviors, mtmc_objects)

Visualization of MTMC Objects in Sequence

This module is used to visualize MTMC objects in sequence mode. The plot() function plots MTMC objects in sequence following the order of timestamps for each matched behavior.

plot(self, global_id: str, output_video_dir: str, frames: Dict[str, Dict[str, Dict[str, Object]]], behaviors: Dict[str, Dict[str, Behavior]], mtmc_objects: Dict[str, Dict[str, Behavior]]) → None
Visualizes MTMC objects in sequence

Parameters

global_id (str) – global ID of an MTMC object

output_video_dir (str) – output directory for videos

frames (Dict[str,Dict[str,Dict[str,Object]]]) – frames extracted from raw data

behaviors (Dict[str,Dict[str,Behavior]]) – behaviors, i.e., tracklets

mtmc_objects (Dict[str,Dict[str,Behavior]]) – MTMC objects, i.e., matches

Returns

None
from mdx.mtmc.viz.mtmc_objects_in_sequence import VizMTMCObjectsInSequence

visualizer = VizMTMCObjectsInSequence(config)
visualizer.plot(global_id, output_video_dir, frames, behaviors, mtmc_objects)

Visualization of MTMC Objects in Top View

This module is used to visualize MTMC objects in topview mode. The plot() function plots MTMC objects’ trajectories on a map from the top view.

plot(self, global_id: str, output_video_dir: str, frames: Dict[str, Dict[str, Dict[str, Object]]], behaviors: Dict[str, Dict[str, Behavior]], mtmc_objects: Dict[str, Dict[str, Behavior]]) → None
Visualizes MTMC objects in top view

Parameters

global_id (str) – global ID of an MTMC object

output_video_dir (str) – output directory for videos

frames (Dict[str,Dict[str,Dict[str,Object]]]) – frames extracted from raw data

behaviors (Dict[str,Dict[str,Behavior]]) – behaviors, i.e., tracklets

mtmc_objects (Dict[str,Dict[str,Behavior]]) – MTMC objects, i.e., matches

Returns

None
from mdx.mtmc.viz.mtmc_objects_in_topview import VizMTMCObjectsInTopview

visualizer = VizMTMCObjectsInTopview(config)
visualizer.plot(global_id, output_video_dir, frames, behaviors, mtmc_objects)

Visualization of Ground-Truth Locations

This module is used to visualize ground-truth locations in top view. The plot() function plots ground-truth trajectories on a map from the top view.

plot(self, output_video_dir: str, ground_truth_locations: Dict[str, Dict[str, List[float]]], max_object_id: int) → None
Visualizes ground-truth locations in top view
Parameters

output_video_dir (str) – output directory for videos

ground_truth_locations (Dict[str,Dict[str,List[float]]]) – ground-truth locations

max_object_id (int) – max object ID

Returns
None
from mdx.mtmc.viz.ground_truth_locations import VizGroundTruthLocations

visualizer = VizGroundTruthLocations(config)
visualizer.plot(output_video_dir, ground_truth_locations, max_object_id)