Connection monitor

This page shows an example on how connector information can be received and processed. More information on monitoring the connection can be found on the connection monitor's documentation page.

Retrieving connector information

In the example below, ConnectorInformation messages are being received for 5 seconds. These messages are then used to determine the average latency and if the connection was healthy in the past 5 seconds.

Python

from openobd import *
import time


# Define how many ConnectorInformation messages should be saved at a time
SAMPLE_SIZE = 5

# Start an openOBD session on a ticket
openobd = OpenOBD()
openobd_session = openobd.start_session_on_ticket(TICKET_ID)    # The ticket ID needs to be filled in (as a string)

# Start the SessionTokenHandler to ensure the openOBD session remains authenticated
SessionTokenHandler(openobd_session)

# Create a ConnectionMonitor object, which will keep receiving and storing ConnectorInformation messages
connection_monitor = ConnectionMonitor(openobd_session, sample_size=SAMPLE_SIZE)

# Wait for ConnectorInformation messages to arrive before retrieving them
time.sleep(SAMPLE_SIZE)
connector_info_list = connection_monitor.get_connector_info_list()

# For each ConnectorInformation message that has been received, check its 'healthy' and 'latency' fields
connection_healthy = True
total_latency = 0
for connector_info in connector_info_list:
    if not connector_info.healthy:
        connection_healthy = False
    total_latency += connector_info.latency

# Print whether the connector had a healthy connection
if connection_healthy:
    print(f"The connection was healthy for the last {SAMPLE_SIZE} seconds.")
else:
    print(f"The connection was not always healthy in the last {SAMPLE_SIZE} seconds.")

# Print the average latency
average_latency = total_latency / len(connector_info_list)
print(f"The connector had an average latency of {average_latency} ms.")

# Stop receiving ConnectorInformation messages
connection_monitor.stop_stream()

# Close the session with a successful result
result = ServiceResult(result=[Result.RESULT_SUCCESS])
openobd_session.finish(result)