Custom Duty Status Report

Create a CSV report of distance driven and time spent in selected HOS duty status types.

This recipe exports HOS log entries for selected duty status types, such as personalConveyance and yardMove, and calculates distance driven for each entry.

import csv
import datetime
import os

import dateutil.tz
import requests

token = os.environ["SAMSARA_API_TOKEN"]
num_days = 1
status_types_of_interest = ["personalConveyance", "yardMove"]

driver_timezones = {}
pagination = {"hasNextPage": True, "endCursor": ""}

while pagination["hasNextPage"]:
    driver_response = requests.request(
        "GET",
        "https://api.samsara.com/fleet/drivers",
        params={"after": pagination["endCursor"]},
        headers={"Authorization": "Bearer " + token},
    ).json()

    for driver in driver_response["data"]:
        driver_timezones[driver["id"]] = driver["timezone"]

    pagination = driver_response["pagination"]

with open("duty_status_report.csv", "w", newline="") as csv_file:
    fieldnames = [
        "Driver",
        "Vehicle",
        "Status",
        "Duration",
        "Distance (miles)",
        "Start Time",
        "End Time",
        "Start Lat/Long",
        "End Lat/Long",
        "Next Status Type",
        "Remark",
    ]
    csv_dict_writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
    csv_dict_writer.writeheader()

    pagination = {"hasNextPage": True, "endCursor": ""}
    end_time = datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")
    start_time = (
        datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=num_days)
    ).isoformat().replace("+00:00", "Z")

    while pagination["hasNextPage"]:
        response = requests.request(
            "GET",
            "https://api.samsara.com/fleet/hos/logs",
            params={
                "startTime": start_time,
                "endTime": end_time,
                "after": pagination["endCursor"],
            },
            headers={"Authorization": "Bearer " + token},
        ).json()

        pagination = response["pagination"]

        for driver_log in response["data"]:
            for i, log_entry in enumerate(driver_log["hosLogs"]):
                if log_entry["hosStatusType"] not in status_types_of_interest:
                    continue

                row = {
                    "Driver": driver_log["driver"]["name"],
                    "Vehicle": log_entry["vehicle"]["id"],
                    "Status": log_entry["hosStatusType"],
                }

                driver_tz = dateutil.tz.gettz(driver_timezones[driver_log["driver"]["id"]])
                log_start_time = datetime.datetime.fromisoformat(
                    log_entry["logStartTime"].replace("Z", "+00:00")
                ).astimezone(driver_tz)
                log_end_time = datetime.datetime.fromisoformat(
                    log_entry["logEndTime"].replace("Z", "+00:00")
                ).astimezone(driver_tz)

                row["Duration"] = log_end_time - log_start_time
                row["Start Time"] = log_start_time
                row["End Time"] = log_end_time
                row["Start Lat/Long"] = (
                    f"{log_entry['logRecordedLocation']['latitude']}/"
                    f"{log_entry['logRecordedLocation']['longitude']}"
                )
                row["Remark"] = log_entry.get("remark", "")

                if log_entry["vehicle"]["id"] != "0":
                    start_odometer_response = requests.request(
                        "GET",
                        "https://api.samsara.com/fleet/vehicles/stats",
                        params={
                            "vehicleIds": log_entry["vehicle"]["id"],
                            "time": log_entry["logStartTime"],
                            "types": "obdOdometerMeters,gpsDistanceMeters",
                        },
                        headers={"Authorization": "Bearer " + token},
                    ).json()

                    end_odometer_response = requests.request(
                        "GET",
                        "https://api.samsara.com/fleet/vehicles/stats",
                        params={
                            "vehicleIds": log_entry["vehicle"]["id"],
                            "time": log_entry["logEndTime"],
                            "types": "obdOdometerMeters,gpsDistanceMeters",
                        },
                        headers={"Authorization": "Bearer " + token},
                    ).json()

                    start_vehicle = start_odometer_response["data"][0]
                    end_vehicle = end_odometer_response["data"][0]

                    if "obdOdometerMeters" in start_vehicle:
                        start_odometer = start_vehicle["obdOdometerMeters"]["value"]
                    else:
                        start_odometer = start_vehicle["gpsDistanceMeters"]["value"]

                    if "obdOdometerMeters" in end_vehicle:
                        end_odometer = end_vehicle["obdOdometerMeters"]["value"]
                    else:
                        end_odometer = end_vehicle["gpsDistanceMeters"]["value"]

                    row["Distance (miles)"] = (end_odometer - start_odometer) / 1609.0

                if i + 1 < len(driver_log["hosLogs"]):
                    next_log_entry = driver_log["hosLogs"][i + 1]
                    row["End Lat/Long"] = (
                        f"{next_log_entry['logRecordedLocation']['latitude']}/"
                        f"{next_log_entry['logRecordedLocation']['longitude']}"
                    )
                    row["Next Status Type"] = next_log_entry["hosStatusType"]

                csv_dict_writer.writerow(row)

How it works

1. Import required packages

   The sample uses Python 3.6 or newer. Install python-dateutil if it is not already available.

2. Set up configuration

   Configure the API token, number of days to report on, and duty status types of interest.

3. Get driver timezones

   The script reads driver timezones so HOS log timestamps can be formatted in each driver’s timezone.

4. Create the CSV report

   Each row represents an HOS log entry that matches a selected duty status type.

5. Get HOS logs

   Call GET /fleet/hos/logs with startTime, endTime, and after to paginate through driver logs.

6. Process each log entry

   For matching entries, the script writes the driver, vehicle, status, duration, start and end times, locations, next status type, and remark.

7. Calculate distance driven

   The script queries vehicle stats at the log start and end times, then subtracts the odometer readings and converts meters to miles.

See Compliance and ELD for more HOS context.