Event-Triggered Compliance Training

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):
    '''
    Fetches Hours of Service (HOS) violations based on a specific time range and tag IDs.

    Args:
        start_time (str): Start time in RFC3339 format.
        end_time (str): End time in RFC3339 format.
        tag_ids (list): List of tag IDs to filter the data.

    Returns:
        list: A list of HOS violations within the specified time range and tags.
    '''
    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') == True:
            has_next_page = True
            after = data['pagination']['endCursor']
        else:
            has_next_page = False

    return hos_violations

def assign_training(driver_ids, course_id, due_date):
    '''
    Assigns a specific training course to the provided drivers with a due date.

    Args:
        driver_ids (list): List of driver IDs to whom the training should be assigned.
        course_id (str): The ID of the course to be assigned.
        due_date (str): The due date for completing the course in RFC3339 format.

    Returns:
        None
    '''
    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):
    '''
    Retrieves a stream of filtered training assignments within a specified time range.

    Args:
        assignment_interval_timestamp (str): Start timestamp in RFC3339 format to
        filter the training assignments.
        course_id (str): The ID of the training course to filter the assignments.

    Returns:
        list: A list of training assignments that match the provided filters.
    '''
    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 generate_rfc3339_timestamp(delta_days):
    '''
    Generates an RFC3339 timestamp by adding or subtracting days from the current date.

    Args:
        delta_days (int): Number of days to add (positive) or subtract (negative) from the
        current date.

    Returns:
        str: The RFC3339 formatted timestamp.
    '''
    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):
    '''
    Generates a timestamp in milliseconds format by adding or subtracting
    days from the current date.

    Args:
        delta_days (int): Number of days to add (positive) or subtract(negative)
        from the current date.

    Returns:
        int: The timestamp in milliseconds.
    '''
    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():
    # Generate RFC3339 timestamps for the HOS violation lookback start and end times
    start_time_rfc3339 = generate_rfc3339_timestamp(HOS_VIOLATION_LOOKBACK_START)
    end_time_rfc3339 = generate_rfc3339_timestamp(HOS_VIOLATION_LOOKBACK_END)

    # Fetch HOS violation events for the specified lookback period and target drivers
    hos_violation_events = get_hos_violations(
        start_time_rfc3339, end_time_rfc3339, TARGET_DRIVER_TAG_IDS
    )

    # Track the count of HOS violations per driver
    driver_hos_violation_count = {}
    for event in hos_violation_events:
        violations = event['violations']
        for violation in violations:
            driver_id = violation['driver']['id']
            if driver_id not in driver_hos_violation_count:
                driver_hos_violation_count[driver_id] = 1
            else:
                driver_hos_violation_count[driver_id] += 1

    # Flag drivers who exceed the HOS violation threshold
    flagged_drivers = {
        driver_id
        for driver_id, count in driver_hos_violation_count.items()
        if count >= HOS_VIOLATION_THRESHOLD
    }

    # Fetch the details of drivers who have already been assigned training
    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'])

    # Identify drivers who are flagged for violations and haven't been assigned training
    drivers_requiring_training = [
        driver for driver in flagged_drivers if driver not in drivers_assigned_training
    ]

    drivers_requiring_training = ','.join(
        [f'driver-{id}' for id in drivers_requiring_training]
    )

    # Generate RFC3339 timestamp for the course completion due date (14 days from now)
    course_due_date_rfc3339 = generate_rfc3339_timestamp(COURSE_COMPLETION_DUE_DAYS)

    # Assign the training to the filtered drivers requiring it
    assign_training(drivers_requiring_training, COURSE_ID, course_due_date_rfc3339)

if __name__ == '__main__':
    main()