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:
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