Scheduling Training

Automate scheduled training course assignments to drivers on a regular cadence using the Samsara Training API.

Overview

This script automates the process of scheduling training course assignments to drivers. It interacts with Samsara APIs to fetch drivers based on tags, check existing assignments, and create new assignments for drivers who have not yet been trained.

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. If you are new to Training, see the Driver Training Guide first.

Example use cases

• An HR manager wants to schedule an “Annual Privacy and Security” training course to be assigned to drivers annually every January.
• A Fleet manager wants to schedule a “Safety Refresher” training course to be assigned to drivers every quarter.

Trigger conditions

• Define the frequency of training (e.g., every 12 months, every quarter)
• Specify when the automation stops
• Specify driver tags for assignment
• Specify conditions under which assignments will skip drivers (e.g., skip assignment if the driver has completed the course within the past 30 days)

Action

• Assign training course with specific due dates

Execution flow

1. Check execution date

   Convert PLANNED_EXECUTION_DATE from string to a datetime object. Compare the current date with the planned execution date and stop the script if they don’t match or if the current year exceeds STOP_YEAR.

2. Generate assignment interval timestamp

   Call generate_rfc3339_timestamp() with TRAINING_LOOKBACK_DAYS to create the assignment lookback period timestamp.

3. Retrieve training assignments

   Call get_training_assignment_details() to fetch training assignments within the defined lookback period for the specified course_id.

4. Identify drivers who have completed training

   Loop through the retrieved assignment_details and identify drivers who have completed the training (status: completed). Add the IDs of completed drivers to drivers_completed_training.

5. Identify drivers who require training

   Loop through the drivers fetched from get_drivers(). If a driver’s ID is not in drivers_completed_training, add them to drivers_requiring_training.

6. Generate training due date

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

7. 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. Ensure your environment has access to Samsara API endpoints and is properly configured for integration. These code snippets can be copied but will need to be customized to match your organization’s specific use case, including correct API tokens and other organizational details.

from datetime import datetime, timedelta
import requests

# API URLs and token
STREAM_TRAINING_ASSIGNMENT_API = 'https://api.samsara.com/training-assignments/stream'
LIST_DRIVERS_API = 'https://api.samsara.com/fleet/drivers'
GET_TRAINING_COURSES_API = 'https://api.samsara.com/training-courses'
ASSIGN_TRAINING_API = 'https://api.samsara.com/training-assignments'
API_KEY = 'Insert your API token'
HEADERS = {'Authorization': f'Bearer {API_KEY}'}

# Constants
COURSE_ID = 'Insert course ID'
TARGET_DRIVER_TAG_IDS = ['Insert the driver tag ID']
TRAINING_LOOKBACK_DAYS = -30
TRAINING_DUE_DAYS = 14
STOP_YEAR = 2030
PLANNED_EXECUTION_DATE = 'Insert date'

def get_drivers(tag_ids):
    drivers = []
    after = ''
    has_next_page = True
    params = {'tagIds': tag_ids}

    while has_next_page:
        params['after'] = after
        response = requests.get(LIST_DRIVERS_API, headers=HEADERS, params=params)
        data = response.json()
        drivers.extend(data.get('data', []))

        if data.get('pagination', {}).get('hasNextPage'):
            has_next_page = True
            after = data['pagination']['endCursor']
        else:
            has_next_page = False

    return drivers

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', ''):
            has_next_page = True
            after = data['pagination']['endCursor']
        else:
            has_next_page = False

    return training_assignments

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

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():
    planned_execution_date = datetime.strptime(PLANNED_EXECUTION_DATE, '%d/%m/%Y').date()
    current_year = datetime.now().year
    current_date = datetime.now().date()

    if current_year > STOP_YEAR or current_date != planned_execution_date:
        return

    drivers_completed_training = []
    drivers_requiring_training = []

    drivers = get_drivers(TARGET_DRIVER_TAG_IDS)
    assignment_interval_rfc3339 = generate_rfc3339_timestamp(TRAINING_LOOKBACK_DAYS)
    assignment_details = get_training_assignment_details(assignment_interval_rfc3339, COURSE_ID)

    for assignment in assignment_details:
        if assignment['learner']['type'] == 'driver' and assignment['status'] == 'completed':
            drivers_completed_training.append(assignment['learner']['id'])
    for driver in drivers:
        if driver['id'] not in drivers_completed_training:
            drivers_requiring_training.append(driver['id'])

    drivers_requiring_training = ','.join([f'driver-{id}' for id in drivers_requiring_training])
    training_due_date_rfc3339 = generate_rfc3339_timestamp(TRAINING_DUE_DAYS)
    assign_training(COURSE_ID, drivers_requiring_training, training_due_date_rfc3339)

if __name__ == '__main__':
    main()

Appendix

Constant definitions

Constant	Type	Description
COURSE_ID	String	ID of the training course that will be assigned.
TARGET_DRIVER_TAG_IDS	List of Strings	List of tag IDs used to filter specific drivers.
TRAINING_LOOKBACK_DAYS	Integer	Number of days to look back for training assignments.
TRAINING_DUE_DAYS	Integer	Number of days from today to set as the due date for training assignments.
STOP_YEAR	Integer	The year after which the automation should stop.
PLANNED_EXECUTION_DATE	String	The date on which the script should be executed.

Function definitions

Function	Description
get_drivers	Retrieves a list of drivers associated with the configured tag_ids, handling pagination.
get_training_assignment_details	Retrieves a stream of training assignments filtered by assignment_interval_timestamp and course_id.
assign_training	Creates training assignments for drivers given course_id, driver_ids, and due_date.
generate_rfc3339_timestamp	Generates a UTC midnight RFC3339 timestamp by adding or subtracting delta_days from the current date. See Timestamps for more information.