Event-Triggered Compliance Training

Automatically assign training courses to drivers based on Hours of Service violations.

Overview

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

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 “Compliance Training” course to a driver when the driver has 2 or more HOS violations in the past 30 days.
• A Fleet manager wants to assign the “Compliance Training” course to a driver when the driver has 3 HOS violations that are marked as “shiftHour” type in the past 45 days.

Trigger conditions

• Specific HOS violation type (e.g., “shiftHours”)
• Specify conditions under which assignments will skip drivers
• Specify driver tags for assignment
• HOS violation count thresholds are exceeded (e.g., 2 violations in 45 days)

Action

• Assign training course with specific due dates

Execution flow

1. Fetch HOS violations

   Convert HOS_VIOLATION_LOOKBACK_START and HOS_VIOLATION_LOOKBACK_END to RFC3339 format. Retrieve HOS violation events that match HOS_VIOLATION_TYPE for the given period and driver tag ID. Count the number of violations for each driver.

2. Filter drivers based on HOS violation threshold

   Identify drivers who have exceeded the HOS_VIOLATION_THRESHOLD. Flag these drivers for training.

3. Identify drivers who require training

   If a driver has already been assigned the course, omit those drivers from drivers_requiring_training.

4. Generate training due date

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

5. 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
HOS_VIOLATION_API = 'https://api.samsara.com/fleet/hos/violations'
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
HOS_VIOLATION_LOOKBACK_START = -45
HOS_VIOLATION_LOOKBACK_END = 0
TARGET_DRIVER_TAG_IDS = ['Insert the driver tag ID']
HOS_VIOLATION_TYPE = 'Insert violation type'
HOS_VIOLATION_THRESHOLD = 3
COURSE_ID = 'Insert course ID'
COURSE_COMPLETION_DUE_DAYS = 14
TRAINING_LOOKBACK_DAYS = -30

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

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 main():
    start_time_rfc3339 = generate_rfc3339_timestamp(HOS_VIOLATION_LOOKBACK_START)
    end_time_rfc3339 = generate_rfc3339_timestamp(HOS_VIOLATION_LOOKBACK_END)

    hos_violation_events = get_hos_violations(start_time_rfc3339, end_time_rfc3339, TARGET_DRIVER_TAG_IDS)

    driver_hos_violation_count = {}
    for event in hos_violation_events:
        for violation in event['violations']:
            driver_id = violation['driver']['id']
            driver_hos_violation_count[driver_id] = driver_hos_violation_count.get(driver_id, 0) + 1

    flagged_drivers = {
        d for d, count in driver_hos_violation_count.items()
        if count >= HOS_VIOLATION_THRESHOLD
    }

    drivers_assigned_training = []
    training_lookback_days_rfc3339 = generate_rfc3339_timestamp(TRAINING_LOOKBACK_DAYS)
    assignment_details = get_training_assignment_details(training_lookback_days_rfc3339, COURSE_ID)
    for assignment in assignment_details:
        if assignment['learner']['type'] == 'driver':
            drivers_assigned_training.append(assignment['learner']['id'])

    drivers_requiring_training = [
        d for d in flagged_drivers if d not in drivers_assigned_training
    ]
    drivers_requiring_training = ','.join([f'driver-{id}' for id in drivers_requiring_training])

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

if __name__ == '__main__':
    main()

Appendix

Constant definitions

Constant	Type	Description
HOS_VIOLATION_LOOKBACK_START	Integer	How many days back to look for HOS violation events.
HOS_VIOLATION_LOOKBACK_END	Integer	End date for looking up HOS violation events.
TARGET_DRIVER_TAG_IDS	List of Strings	Tag IDs used to filter specific drivers.
HOS_VIOLATION_TYPE	String	Specific HOS violation type to track.
HOS_VIOLATION_THRESHOLD	Integer	Threshold of HOS violations before a driver is flagged.
COURSE_ID	String	ID of the training course that will be assigned.
COURSE_COMPLETION_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_hos_violations	Fetches HOS violation events for a given time range, tag IDs, and violation type.
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.