ObjectTracking¶
- class CorrelationTracker(max_objects=None, deregister_frames=30, max_distance=50, min_inertia=5, confidence_threshold=0.0, history_length=0, enter_cb=None, exit_cb=None)¶
Bases:
TrackerAlgorithm
[TrackableCorrelationPrediction
,ObjectDetectionPrediction
]Track objects based on a correlation tracking algorithm.
Typical usage:
def object_enters(object_id, prediction): print("{}: {} enters".format(object_id, prediction.label)) def object_exits(object_id, prediction): print("{} exits".format(prediction.label)) obj_detect = edgeiq.ObjectDetection( 'alwaysai/res10_300x300_ssd_iter_140000') obj_detect.load(engine=edgeiq.Engine.DNN) tracker = edgeiq.CorrelationTracker( deregister_frames=20, max_distance=50, enter_cb=object_enters, exit_cb=object_exits) while True: <get video frame> results = obj_detect.detect_objects(frame) objects = tracker.update(results.predictions, frame) # Use the object dictionary to create a new prediction list for (object_id, prediction) in objects.items(): new_label = 'object {}'.format(object_id) prediction.label = new_label predictions.append(prediction) frame = edgeiq.markup_image(frame, predictions)
- Parameters
max_objects (
Optional
[int
]) – The maximum number of objects to track.deregister_frames (
int
) – The number of frames before deregistering an object that can no longer be found.max_distance (
int
) – The maximum distance between two centroids to associate an object.min_inertia (
int
) – The inertia is computed by incrementing when a detection is matched and decrementing when it isn’t, flooring at zero. This value is the threshold at which an item begins being treated as a tracked object.confidence_threshold (
float
) – A threshold used in two ways: First, to split the objects into a group of high confidence and one of low confidence to perform priority matching on the high confidence predictions. Then, any unmatched predictions must have a confidence above the threshold to begin being tracked.history_length (
int
) – The number of historical predictions to remember.enter_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time a new object is detected.exit_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time an existing object is removed from the tracking list. This event occurs deregister_frames after the object is last detected.
- update(predictions, image)¶
Update tracked objects based on a new set of predictions and a new image.
- Parameters
predictions (
Sequence
[ObjectDetectionPrediction
]) – The list of bounding boxes to track.- Return type
- Returns
A dictionary with object ID as the key and the
ObjectDetectionPrediction
as the value.
- publish_analytics(results, tag=None, **kwargs)¶
Publish Object Tracking results to the alwaysAI Analytics Service
Example usage:
try: tracker.publish_analytics(results, tag='custom_tag') except edgeiq.PublishError as e: # Retry publish except edgeiq.ConnectionError as e: # Save state and exit app to reconnect
- Parameters
results (
TrackingResults
) – The results to publish.tag (
Optional
[Any
]) – Additional information to assist in querying and visualizations.
- Raises
ConnectionBlockedError
when using connection to the alwaysAI Device Agent and resources are at capacity,- Raises
PacketRateError
when publish rate exceeds current limit,- Raises
PacketSizeError
when packet size exceeds current limit. Packet publish size and rate limits will be provided in the error message.
- remove_all()¶
Remove all tracked objects from the list.
The exit handler will be called if the tracked object has met min inertia.
- Parameters
id – The tracking id of the object to be deleted from the list of objects that are being tracked.
- remove_id(id)¶
Remove a particular object from the list of objects being tracked
The function removes the object if a tracked object id is provided. Otherwise, the function will not update anything. The exit handler will be called if the tracked object has met min inertia.
- Parameters
id (
int
) – The tracking id of the object to be deleted from the list of objects that are being tracked.
- class TrackableCorrelationPrediction(*args, **kwargs)¶
Bases:
TrackablePrediction
[ObjectDetectionPrediction
]- handle_found(prediction, dereg_tracked_obj, **kwargs)¶
- handle_disappeared(image, reg_tracked_obj, can_track_new_obj, **kwargs)¶
- property box: BoundingBox¶
The bounding box around the object.
- Return type
- property confidence: float¶
The confidence of this prediction.
- Return type
float
- handle_removed()¶
- property history: List[PredictionT]¶
- Return type
List
[TypeVar
(PredictionT
, bound=ObjectDetectionPrediction
)]
- property index: int¶
The index of this result in the master label list.
- Return type
int
- property is_initialized: bool¶
- Return type
bool
- property is_lost: bool¶
- Return type
bool
- property label: str¶
The label describing this prediction result.
- Return type
str
- step(**trackable_kwargs)¶
- property tid: int¶
- Return type
int
- tracker_id = 0¶
- tracker_init_id = 0¶
- class CentroidTracker(deregister_frames=30, max_distance=50, min_inertia=5, confidence_threshold=0.0, history_length=0, enter_cb=None, exit_cb=None)¶
Bases:
TrackerAlgorithm
[TrackablePrediction
[ObjectDetectionPrediction
],ObjectDetectionPrediction
]Associate a bounding box with an object ID based on distances from previous detections.
Typical usage:
def object_enters(object_id, prediction): print("{}: {} enters".format(object_id, prediction.label)) def object_exits(object_id, prediction): print("{} exits".format(prediction.label)) obj_detect = edgeiq.ObjectDetection( 'alwaysai/res10_300x300_ssd_iter_140000' ) obj_detect.load(engine=edgeiq.Engine.DNN) centroid_tracker = edgeiq.CentroidTracker( deregister_frames=20, max_distance=50, enter_cb=object_enters, exit_cb=object_exits ) while True: <get video frame> results = obj_detect.detect_objects(frame) objects = centroid_tracker.update(results.predictions) # Use the object dictionary to create a new prediction list for (object_id, prediction) in objects.items(): new_label = 'object {}'.format(object_id) prediction.label = new_label predictions.append(prediction) frame = edgeiq.markup_image(frame, predictions)
- Parameters
deregister_frames (
int
) – The number of frames before deregistering an object that can no longer be found.max_distance (
int
) – The maximum distance between two centroids to associate an object.min_inertia (
int
) – The inertia is computed by incrementing when a detection is matched and decrementing when it isn’t, flooring at zero. This value is the threshold at which an item begins being treated as a tracked object.confidence_threshold (
float
) – The minimum confidence level for starting to track a new object.history_length (
int
) – The number of historical predictions to remember.enter_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time a new object is detected.exit_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time an existing object is removed from the tracking list. This event occurs deregister_frames after the object is last detected.
- publish_analytics(results, tag=None, **kwargs)¶
Publish Object Tracking results to the alwaysAI Analytics Service
Example usage:
try: tracker.publish_analytics(results, tag='custom_tag') except edgeiq.PublishError as e: # Retry publish except edgeiq.ConnectionError as e: # Save state and exit app to reconnect
- Parameters
results (
TrackingResults
) – The results to publish.tag (
Optional
[Any
]) – Additional information to assist in querying and visualizations.
- Raises
ConnectionBlockedError
when using connection to the alwaysAI Device Agent and resources are at capacity,- Raises
PacketRateError
when publish rate exceeds current limit,- Raises
PacketSizeError
when packet size exceeds current limit. Packet publish size and rate limits will be provided in the error message.
- remove_all()¶
Remove all tracked objects from the list.
The exit handler will be called if the tracked object has met min inertia.
- Parameters
id – The tracking id of the object to be deleted from the list of objects that are being tracked.
- remove_id(id)¶
Remove a particular object from the list of objects being tracked
The function removes the object if a tracked object id is provided. Otherwise, the function will not update anything. The exit handler will be called if the tracked object has met min inertia.
- Parameters
id (
int
) – The tracking id of the object to be deleted from the list of objects that are being tracked.
- update(predictions, **trackable_kwargs)¶
Update tracked objects based on a new set of predictions.
Performs the following steps:
Match predictions with initialized tracked objects (those meeting the min inertia)
Match unmatched predictions from (1) to uninitialized tracked objects.
Start tracking any remaining unmatched predictions that meet the minimum confidence threshold.
- Parameters
predictions (
Sequence
[TypeVar
(PredictionT
, bound=ObjectDetectionPrediction
)]) – The list of bounding boxes to track.- Return type
TrackingResults
[TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)]
- class KalmanTracker(deregister_frames=30, max_distance=50, min_inertia=5, confidence_threshold=0.0, history_length=0, enter_cb=None, exit_cb=None)¶
Bases:
TrackerAlgorithm
[TrackableKalmanPrediction
,ObjectDetectionPrediction
]Associate a bounding box with an object ID based on distances from previous detections. Each tracked object is assigned a Kalman Filter, which is updated after each frame, whose purpose is to model and understand the motion of the object being tracked. This adds a level of robustness to the tracking system, as the future position/velocity of the object may be estimated and used when matching objects in frame rather than just the object’s current position. This additional complexity adds computational overhead.
The output predictions have additional attributes:
estimated_position: The estimated position from the Kalman filter
estimated_velocity: The estimated velocity from the Kalman filter
Typical usage:
def object_enters(object_id, prediction): print("{}: {} enters".format(object_id, prediction.label)) def object_exits(object_id, prediction): print("{} exits".format(prediction.label)) obj_detect = edgeiq.ObjectDetection( 'alwaysai/res10_300x300_ssd_iter_140000') obj_detect.load(engine=edgeiq.Engine.DNN) kalman_tracker = edgeiq.KalmanTracker( deregister_frames=20, max_distance=50, enter_cb=object_enters, exit_cb=object_exits) while True: <get video frame> results = obj_detect.detect_objects(frame) objects = kalman_tracker.update(results.predictions) # Use the object dictionary to create a new prediction list for (object_id, prediction) in objects.items(): new_label = 'object {}: position {} velocity {}'.format( object_id, prediction.estimated_position, prediction.estimated_velocity) prediction.label = new_label predictions.append(prediction) frame = edgeiq.markup_image(frame, predictions)
- Parameters
deregister_frames (
int
) – The number of frames before deregistering an object that can no longer be found.max_distance (
int
) – The maximum distance between two centroids to associate an object.min_inertia (
int
) – The inertia is computed by incrementing when a detection is matched and decrementing when it isn’t, flooring at zero. This value is the threshold at which an item begins being treated as a tracked object.confidence_threshold (
float
) – A threshold used in two ways: First, to split the objects into a group of high confidence and one of low confidence to perform priority matching on the high confidence predictions. Then, any unmatched predictions must have a confidence above the threshold to begin being tracked.history_length (
int
) – The number of historical predictions to remember.enter_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time a new object is detected.exit_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time an existing object is removed from the tracking list. This event occurs deregister_frames after the object is last detected.
- publish_analytics(results, tag=None, **kwargs)¶
Publish Object Tracking results to the alwaysAI Analytics Service
Example usage:
try: tracker.publish_analytics(results, tag='custom_tag') except edgeiq.PublishError as e: # Retry publish except edgeiq.ConnectionError as e: # Save state and exit app to reconnect
- Parameters
results (
TrackingResults
) – The results to publish.tag (
Optional
[Any
]) – Additional information to assist in querying and visualizations.
- Raises
ConnectionBlockedError
when using connection to the alwaysAI Device Agent and resources are at capacity,- Raises
PacketRateError
when publish rate exceeds current limit,- Raises
PacketSizeError
when packet size exceeds current limit. Packet publish size and rate limits will be provided in the error message.
- remove_all()¶
Remove all tracked objects from the list.
The exit handler will be called if the tracked object has met min inertia.
- Parameters
id – The tracking id of the object to be deleted from the list of objects that are being tracked.
- remove_id(id)¶
Remove a particular object from the list of objects being tracked
The function removes the object if a tracked object id is provided. Otherwise, the function will not update anything. The exit handler will be called if the tracked object has met min inertia.
- Parameters
id (
int
) – The tracking id of the object to be deleted from the list of objects that are being tracked.
- update(predictions, **trackable_kwargs)¶
Update tracked objects based on a new set of predictions.
Performs the following steps:
Match predictions with initialized tracked objects (those meeting the min inertia)
Match unmatched predictions from (1) to uninitialized tracked objects.
Start tracking any remaining unmatched predictions that meet the minimum confidence threshold.
- Parameters
predictions (
Sequence
[TypeVar
(PredictionT
, bound=ObjectDetectionPrediction
)]) – The list of bounding boxes to track.- Return type
TrackingResults
[TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)]
- class TrackableKalmanPrediction(*args, **kwargs)¶
Bases:
TrackablePrediction
[ObjectDetectionPrediction
]- property estimated_position: ndarray¶
- Return type
ndarray
- property estimated_velocity: ndarray¶
- Return type
ndarray
- step(**kwargs)¶
- handle_found(prediction, **kwargs)¶
- property box: BoundingBox¶
The bounding box around the object.
- Return type
- property confidence: float¶
The confidence of this prediction.
- Return type
float
- handle_disappeared(**trackable_kwargs)¶
- handle_removed()¶
- property history: List[PredictionT]¶
- Return type
List
[TypeVar
(PredictionT
, bound=ObjectDetectionPrediction
)]
- property index: int¶
The index of this result in the master label list.
- Return type
int
- property is_initialized: bool¶
- Return type
bool
- property is_lost: bool¶
- Return type
bool
- property label: str¶
The label describing this prediction result.
- Return type
str
- property tid: int¶
- Return type
int
- tracker_id = 0¶
- tracker_init_id = 0¶
- class TrackingResults(objects, tracking_algorithm)¶
Bases:
Dict
[int
,TrackablePredictionT
]The output results of the tracker
- property tracking_algorithm¶
The tracking algorithm used
- clear() None. Remove all items from D. ¶
- copy() a shallow copy of D ¶
- fromkeys(value=None, /)¶
Create a new dictionary with keys from iterable and values set to value.
- get(key, default=None, /)¶
Return the value for key if key is in the dictionary, else default.
- items() a set-like object providing a view on D's items ¶
- keys() a set-like object providing a view on D's keys ¶
- pop(k[, d]) v, remove specified key and return the corresponding value. ¶
If the key is not found, return the default if given; otherwise, raise a KeyError.
- popitem()¶
Remove and return a (key, value) pair as a 2-tuple.
Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.
- setdefault(key, default=None, /)¶
Insert key with a value of default if key is not in the dictionary.
Return the value for key if key is in the dictionary, else default.
- update([E, ]**F) None. Update D from dict/iterable E and F. ¶
If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]
- values() an object providing a view on D's values ¶
- class TrackerAnalytics(annotations)¶
Bases:
TrackerAlgorithm
[TrackablePrediction
[ObjectDetectionPrediction
],ObjectDetectionPrediction
]Reads an analytics file and returns result with similar interface as TrackerAlgorithm.
- Parameters
annotations (
List
[TrackingResults
]) – List of tracking results from a single stream.
Typical usage:
# get tracker results from annotation file annotation_file = 'analytics.txt' annotation_results = edgeiq.analytics_services.load_analytics_results(annotation_file) tracker = edgeiq.TrackerAnalytics(annotations=annotation_results) results = tracker.update(predictions)
- update(predictions, **trackable_kwargs)¶
Update tracked objects based on a new set of predictions.
Performs the following steps:
Match predictions with initialized tracked objects (those meeting the min inertia)
Match unmatched predictions from (1) to uninitialized tracked objects.
Start tracking any remaining unmatched predictions that meet the minimum confidence threshold.
- Parameters
predictions (
Sequence
[ObjectDetectionPrediction
]) – The list of bounding boxes to track.- Return type
- publish_analytics(results, tag=None, **kwargs)¶
Publish Object Tracking results to the alwaysAI Analytics Service
Example usage:
try: tracker.publish_analytics(results, tag='custom_tag') except edgeiq.PublishError as e: # Retry publish except edgeiq.ConnectionError as e: # Save state and exit app to reconnect
- Parameters
results (
TrackingResults
) – The results to publish.tag (
Optional
[Any
]) – Additional information to assist in querying and visualizations.
- Raises
ConnectionBlockedError
when using connection to the alwaysAI Device Agent and resources are at capacity,- Raises
PacketRateError
when publish rate exceeds current limit,- Raises
PacketSizeError
when packet size exceeds current limit. Packet publish size and rate limits will be provided in the error message.
- remove_all()¶
Remove all tracked objects from the list.
The exit handler will be called if the tracked object has met min inertia.
- Parameters
id – The tracking id of the object to be deleted from the list of objects that are being tracked.
- remove_id(id)¶
Remove a particular object from the list of objects being tracked
The function removes the object if a tracked object id is provided. Otherwise, the function will not update anything. The exit handler will be called if the tracked object has met min inertia.
- Parameters
id (
int
) – The tracking id of the object to be deleted from the list of objects that are being tracked.
- class TrackablePrediction(prediction, deregister_frames, min_inertia, history_length, enter_cb=None, exit_cb=None, id=None, initialized=False)¶
Bases:
Generic
[PredictionT
],ObjectDetectionPrediction
The wrapper class for maintaining tracker state for a prediction.
This class inherits its interface from
edgeiq.ObjectDetectionPrediction
and is generically parameterized by the prediction input parameter which can be any subclass ofObjectDetectionPrediction
.- tracker_init_id = 0¶
- tracker_id = 0¶
- step(**trackable_kwargs)¶
- handle_found(prediction, **trackable_kwargs)¶
- handle_disappeared(**trackable_kwargs)¶
- handle_removed()¶
- property tid: int¶
- Return type
int
- property is_initialized: bool¶
- Return type
bool
- property is_lost: bool¶
- Return type
bool
- property history: List[PredictionT]¶
- Return type
List
[TypeVar
(PredictionT
, bound=ObjectDetectionPrediction
)]
- property label: str¶
The label describing this prediction result.
- Return type
str
- property index: int¶
The index of this result in the master label list.
- Return type
int
- property box: BoundingBox¶
The bounding box around the object.
- Return type
- property confidence: float¶
The confidence of this prediction.
- Return type
float
- class TrackerAlgorithm(deregister_frames, min_inertia, confidence_threshold, history_length, enter_cb, exit_cb, trackable, distance_functions, match_optimizer)¶
Bases:
Generic
[TrackablePredictionT
,PredictionT
]Core Object Tracking algorithm which takes a list of predictions and tracks them over frames, outputting a list of TrackingResults after each update.
This class is generically parameterized by the trackable prediction type is defined by the trackable input parameter and determines the output of the
update()
function. The trackable prediction type can be any subclass ofedgeiq.object_tracking.trackable_prediction.TrackablePrediction
.- Parameters
deregister_frames (
int
) – The number of frames before deregistering an object that can no longer be found.min_inertia (
int
) – The inertia is computed by incrementing when a detection is matched and decrementing when it isn’t, flooring at zero. This value is the threshold at which an item begins being treated as a tracked object.confidence_threshold (
float
) – A threshold used in two ways: First, to split the objects into a group of high confidence and one of low confidence to perform priority matching on the high confidence predictions. Then, any unmatched predictions must have a confidence above the threshold to begin being tracked.history_length (
int
) – The number of historical predictions to remember.enter_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time a new object is detected.exit_cb (
Optional
[Callable
[[int
,TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)],None
]]) – A callback function to be called each time an existing object is removed from the tracking list. This event occurs deregister_frames after the object is last detected.trackable (
Type
[TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)]) – The class to be instantiated for each tracked object.distance_functions (
List
[Callable
[[TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
),TypeVar
(PredictionT
, bound=ObjectDetectionPrediction
)],float
]]) – A list of functions to use to determine distance used for matching. The functions are called in the list order, so that the first function will perform matching first, the unmatched will be passed to the second function, and so on.match_optimizer (
Callable
[[ndarray
],List
[Tuple
[int
,int
]]]) – A function used to determine the optimal match based in the distances from the distance function.
- update(predictions, **trackable_kwargs)¶
Update tracked objects based on a new set of predictions.
Performs the following steps:
Match predictions with initialized tracked objects (those meeting the min inertia)
Match unmatched predictions from (1) to uninitialized tracked objects.
Start tracking any remaining unmatched predictions that meet the minimum confidence threshold.
- Parameters
predictions (
Sequence
[TypeVar
(PredictionT
, bound=ObjectDetectionPrediction
)]) – The list of bounding boxes to track.- Return type
TrackingResults
[TypeVar
(TrackablePredictionT
, bound=TrackablePrediction
)]
- remove_id(id)¶
Remove a particular object from the list of objects being tracked
The function removes the object if a tracked object id is provided. Otherwise, the function will not update anything. The exit handler will be called if the tracked object has met min inertia.
- Parameters
id (
int
) – The tracking id of the object to be deleted from the list of objects that are being tracked.
- remove_all()¶
Remove all tracked objects from the list.
The exit handler will be called if the tracked object has met min inertia.
- Parameters
id – The tracking id of the object to be deleted from the list of objects that are being tracked.
- publish_analytics(results, tag=None, **kwargs)¶
Publish Object Tracking results to the alwaysAI Analytics Service
Example usage:
try: tracker.publish_analytics(results, tag='custom_tag') except edgeiq.PublishError as e: # Retry publish except edgeiq.ConnectionError as e: # Save state and exit app to reconnect
- Parameters
results (
TrackingResults
) – The results to publish.tag (
Optional
[Any
]) – Additional information to assist in querying and visualizations.
- Raises
ConnectionBlockedError
when using connection to the alwaysAI Device Agent and resources are at capacity,- Raises
PacketRateError
when publish rate exceeds current limit,- Raises
PacketSizeError
when packet size exceeds current limit. Packet publish size and rate limits will be provided in the error message.
- match_greedy(dist)¶
- Return type
List
[Tuple
[int
,int
]]
- match_optimal(dist)¶
- Return type
List
[Tuple
[int
,int
]]