Analytics¶

class AnalyticsPacket(type, tag, qos, results)¶

Attributes added to all analytics packets loaded from file.

type: str¶

tag: Optional[Any]¶

qos: int¶

results: Any¶

class CustomEvent(results, tag=None, qos=0)¶

A custom event published by edgeiq.publish_analytics().

Parameters: results (Any) – The custom event data parsed from the analytics file.

results: Any¶

tag: Any = None¶

qos: int = 0¶

load_analytics_results(filepath, packet_types=None, num_logs=None)¶

Load results from file published by the alwaysAI Analytics Service.

Typical usage:

deserialized_results = edgeiq.load_analytics_results('logs/analytics.txt')
left_camera_results = [result for result in deserialized_results if 'left' in result.tag]
right_camera_results = [result for result in deserialized_results if 'right' in result.tag]

Parameters

filepath (str) – The full path to the file to load.
packet_types (Optional[List[str]]) – The list of packet types to filter by. None filters for all packet types.
num_logs (Optional[int]) – The number of logs to read from the end of the file. None reads all lines in the file.

Return type

Sequence[Union[AnalyticsPacket, CustomEvent, TrackingResults, ClassificationResults, HumanPoseResult, ObjectDetectionResults, ReIdentificationResults, InstanceSegmentationResults, BarcodeDetectionResults, QRCodeDetectionResults, OccurrenceEvent, ValueEvent, StartTimedEvent, EndTimedEvent]]

Returns

A list of the deserialized results. Each deserialized result will include a tag property.

parse_analytics_packet(packet_str)¶

Takes in an analytics packet as input and parses it to extract relevant information. The analytics packet is expected to be in JSON format. The function then returns the extracted result object.

Typical Usage:

packet = '{"type": "ObjectDetectionResult", "results": {...}}'
result = parse_analytics_packet(packet)

Return type: Union[AnalyticsPacket, CustomEvent, TrackingResults, ClassificationResults, HumanPoseResult, ObjectDetectionResults, ReIdentificationResults, InstanceSegmentationResults, BarcodeDetectionResults, QRCodeDetectionResults, OccurrenceEvent, ValueEvent, StartTimedEvent, EndTimedEvent]

publish_analytics(results, tag=None, **kwargs)¶

Publish data to the alwaysAI Analytics Service

Example usage:

try:
    edgeiq.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 (Any) – The results to publish. Must be JSON-serializable
tag (Optional[Any]) – Additional information to assist in querying and visualizations. Must be JSON serializable.

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.

write_object_detection_results_to_analytics_file(output_file_path, results, tag=None)¶

Write results to an analytics file at a custom path.

This is useful for writing to analytics files in a scripting environment, as opposed to an application environment. This function doesn’t require alwaysai.app.json.

Parameters

output_file_path (str) – Full path to the analytics file to write to. The file may not exist, but the directory must exist.
results (ObjectDetectionResults) – The ObjectDetectionResults to write to the analytics file.

Return type

None

write_tracking_results_to_analytics_file(output_file_path, results, tag=None)¶

Write tracking results to an analytics file at a custom path.

This is useful for writing to analytics files in a scripting environment, as opposed to an application environment. This function doesn’t require alwaysai.app.json.

Parameters

output_file_path (str) – Full path to the analytics file to write to. The file may not exist, but the directory must exist.
results (TrackingResults) – The TrackingResults to write to the analytics file.

Return type

None