Event-Triggered Safety Training

Automatically assign training courses to drivers based on safety events and safety scores.

Overview

This script automates the process of assigning training courses to drivers based on their safety events. It interacts with Samsara APIs to fetch drivers based on tags, check existing assignments, and create new assignments for drivers based on their safety events.

This guide assumes you are already familiar with the basics of the Samsara Training Product and the overall GET Training Assignments, POST Training Assignments, and GET Training Courses APIs.

Example use cases

• A Safety manager wants to assign the “Electronic Distractions” course to a driver when the driver has 2 or more Mobile Usage events in the past 30 days.
• A Fleet manager wants to assign the “Speeding” course to a driver when the driver has 3 Severe Speeding events that are marked as “Needs Coaching” status in the past 45 days AND has a safety score below 80.

Trigger conditions

• Specify safety event count threshold (e.g., 2 events in 45 days)
• Specify safety behavior (e.g., “Mobile Usage”)
• Specify safety event status (e.g., “Needs Coaching”)
• Specify conditions under which assignments will skip drivers
• Specify driver tags for assignment

Action

• Assign training course with specific due dates

Execution flow

1. Fetch safety events

   Convert SAFETY_EVENTS_LOOKBACK_START and SAFETY_EVENTS_LOOKBACK_END to RFC3339 format. Retrieve safety events for the given period and tag IDs.

2. Filter drivers based on event behavior and status

   Count the number of times each driver has exhibited the specified safety behavior, considering only the events for which the status matches EVENT_STATUS. Filter only drivers who meet or exceed BEHAVIOUR_THRESHOLD.

3. Check driver safety scores

   Convert SAFETY_EVENTS_LOOKBACK_START and SAFETY_EVENTS_LOOKBACK_END to milliseconds. Retrieve safety scores for each driver and filter drivers whose score is below SAFETY_SCORE_THRESHOLD.

4. Identify drivers who require training

   Loop through the drivers that meet the above thresholds. If a driver has already been assigned the course, omit those drivers from drivers_requiring_training.

5. Generate training due date

   Call generate_rfc3339_timestamp() with TRAINING_DUE_DAYS to get the training due date in RFC3339 format.

6. Assign training

   Call assign_training() to assign the course to drivers in drivers_requiring_training with the generated training due date.

Script

To run certain automations, you will need to set up a development environment. These code snippets can be copied but will need to be customized to match your organization’s specific use case.

from datetime import datetime, timedelta
import requests

# API URLs and token
SAFETY_EVENTS_API = 'https://api.samsara.com/fleet/safety-events'
DRIVER_SAFETY_SCORE_API = 'https://api.samsara.com/v1/fleet/drivers/{}/safety/score'
ASSIGN_TRAINING_API = 'https://api.samsara.com/training-assignments'
STREAM_TRAINING_ASSIGNMENT_API = 'https://api.samsara.com/training-assignments/stream'
API_TOKEN = 'Insert your API token'
HEADERS = {'Accept': 'application/json', 'Authorization': f'Bearer {API_TOKEN}'}

# Constants
SAFETY_EVENTS_LOOKBACK_START = -45
SAFETY_EVENTS_LOOKBACK_END = 0
BEHAVIOR_TO_COACH = 'Insert behavior here'
EVENT_STATUS = 'Insert event status here'
TARGET_DRIVER_TAG_IDS = ['Insert the driver tag ID']
SAFETY_VIOLATION_THRESHOLD = 2
SAFETY_SCORE_THRESHOLD = 80
COURSE_ID = 'Insert course ID'
TRAINING_DUE_DAYS = 14
TRAINING_LOOKBACK_DAYS = -30

def get_safety_events(start_time, end_time, tag_ids):
    params = {'startTime': start_time, 'endTime': end_time, 'tagIds': ','.join(tag_ids)}
    safety_events = []
    after = ''
    has_next_page = True
    while has_next_page:
        params['after'] = after
        response = requests.get(SAFETY_EVENTS_API, headers=HEADERS, params=params)
        data = response.json()
        safety_events.extend(data['data'])
        if data.get('pagination', {}).get('hasNextPage'):
            after = data['pagination']['endCursor']
        else:
            has_next_page = False
    return safety_events

def get_driver_safety_score(driver_id, start_ms, end_ms):
    URL = f'{DRIVER_SAFETY_SCORE_API.format(driver_id)}'
    params = {'startMs': start_ms, 'endMs': end_ms}
    response = requests.get(URL, headers=HEADERS, params=params)
    return response.json()

def assign_training(driver_ids, course_id, due_date):
    params = {'learnerIds': driver_ids, 'courseId': course_id, 'dueAtTime': due_date}
    requests.post(ASSIGN_TRAINING_API, headers=HEADERS, params=params)

def get_training_assignment_details(assignment_interval_timestamp, course_id):
    training_assignments = []
    after = ''
    params = {'startTime': assignment_interval_timestamp, 'courseIds': [course_id]}
    has_next_page = True
    while has_next_page:
        params['after'] = after
        response = requests.get(STREAM_TRAINING_ASSIGNMENT_API, headers=HEADERS, params=params)
        data = response.json()
        training_assignments.extend(data['data'])
        if data.get('pagination', {}).get('hasNextPage', ''):
            after = data['pagination']['endCursor']
        else:
            has_next_page = False
    return training_assignments

def generate_rfc3339_timestamp(delta_days):
    target_date = (datetime.now() + timedelta(days=delta_days)).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    return target_date.strftime('%Y-%m-%dT%H:%M:%SZ')

def generate_milliseconds_timestamp(delta_days):
    target_date = (datetime.now() + timedelta(days=delta_days)).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    return int(target_date.timestamp()) * 1000

def main():
    start_time_rfc3339 = generate_rfc3339_timestamp(SAFETY_EVENTS_LOOKBACK_START)
    end_time_rfc3339 = generate_rfc3339_timestamp(SAFETY_EVENTS_LOOKBACK_END)
    start_time_ms = generate_milliseconds_timestamp(SAFETY_EVENTS_LOOKBACK_START)
    end_time_ms = generate_milliseconds_timestamp(SAFETY_EVENTS_LOOKBACK_END)

    safety_events = get_safety_events(start_time_rfc3339, end_time_rfc3339, TARGET_DRIVER_TAG_IDS)

    driver_event_behavior_count = {}
    for event in safety_events:
        driver_id = event['driver']['id']
        behavior_label = event['behaviorLabels'][0]['name']
        coaching_state = event['coachingState']
        if behavior_label == BEHAVIOR_TO_COACH and coaching_state == EVENT_STATUS:
            driver_event_behavior_count[driver_id] = driver_event_behavior_count.get(driver_id, 0) + 1

    flagged_drivers = {
        driver_id for driver_id, count in driver_event_behavior_count.items()
        if count >= SAFETY_VIOLATION_THRESHOLD
    }

    drivers_with_violation_record = []
    for driver_id in flagged_drivers:
        safety_score_data = get_driver_safety_score(driver_id, start_time_ms, end_time_ms)
        if safety_score_data and safety_score_data.get('safetyScore') < SAFETY_SCORE_THRESHOLD:
            drivers_with_violation_record.append(driver_id)

    training_lookback_days_rfc3339 = generate_rfc3339_timestamp(TRAINING_LOOKBACK_DAYS)
    assignment_details = get_training_assignment_details(training_lookback_days_rfc3339, COURSE_ID)

    drivers_with_existing_training = [
        a['learner']['id'] for a in assignment_details if a['learner']['type'] == 'driver'
    ]
    drivers_requiring_training = [
        d for d in flagged_drivers if d not in drivers_with_existing_training
    ]
    drivers_requiring_training = ','.join([f'driver-{id}' for id in drivers_requiring_training])

    course_due_date_rfc3339 = generate_rfc3339_timestamp(TRAINING_DUE_DAYS)
    assign_training(drivers_requiring_training, COURSE_ID, course_due_date_rfc3339)

if __name__ == '__main__':
    main()

Appendix

Constant definitions

Constant	Type	Description
SAFETY_EVENTS_LOOKBACK_START	Integer	How many days back to look for safety events.
SAFETY_EVENTS_LOOKBACK_END	Integer	End date for looking up safety events.
BEHAVIOR_TO_COACH	String	Specific safety event behavior to track.
EVENT_STATUS	String	The status for a safety event.
TARGET_DRIVER_TAG_IDS	List of Strings	Tag IDs used to filter specific drivers.
SAFETY_VIOLATION_THRESHOLD	Integer	Threshold for number of behaviors before training is assigned.
SAFETY_SCORE_THRESHOLD	Integer	Drivers with a safety score below this threshold will be considered.
COURSE_ID	String	ID of the training course that will be assigned.
TRAINING_DUE_DAYS	Integer	Number of days from today to set as the due date.
TRAINING_LOOKBACK_DAYS	Integer	Number of days to look back for training assignments.

Function definitions

Function	Description
get_safety_events	Fetches safety events for a given time range and filters by driver tag IDs.
get_driver_safety_score	Retrieves a driver’s safety score for a given time range.
assign_training	Creates training assignments for drivers.
get_training_assignment_details	Retrieves training assignments filtered by assignment_interval_timestamp and course_id.
generate_rfc3339_timestamp	Generates a UTC midnight RFC3339 timestamp. See Timestamps.
generate_milliseconds_timestamp	Generates a millisecond timestamp by adding or subtracting delta_days from now.