Fleet Telemetry with Amazon Timestream

Build a supply chain analytics backend using Amazon Timestream for InfluxDB.

Introduction

30 Second Summary

Every time you track a package from warehouse to doorstep, a delivery company is juggling thousands of data points per vehicle. Speed, fuel, location, idle time. Without the right database, all that data becomes noise instead of insight.

In this project, you will build a fleet analytics backend for a fictional last-mile delivery company called SwiftRoute Logistics. You will simulate 10 delivery vans generating telemetry, ingest that data into Amazon Timestream for InfluxDB, and run time-series queries that answer real supply chain business questions.

What You'll Build

A Python-based fleet telemetry pipeline that simulates delivery vehicles, writes data into a time-series database, and queries it for actionable logistics insights.

By the end of this project, you'll have:

  • A fleet simulator that generates realistic GPS, speed, fuel, and delivery status data for 10 vehicles across 4 hours of routes.
  • A data ingestion pipeline that writes approximately 4800 telemetry points to Amazon Timestream for InfluxDB using the Python client.
  • A set of Flux queries that answer real logistics business questions including SLA breach detection, fuel efficiency rankings, and idle time analysis.
  • Secret Mission: Add delivery throughput analysis with staffing gap detection across time windows.

Are there any prerequisites?

You will need an AWS account with billing enabled, Python 3 installed on your machine, and a code editor like VS Code or Cursor. Basic Python knowledge is helpful but not required.

Before We Start

Before We Start...

Set Up Your Environment and Create the Timestream Instance

SwiftRoute's delivery vans generate a constant stream of GPS coordinates, speed readings, and fuel levels. To store and query this time-series telemetry, you need a database designed specifically for timestamped data.

In this step, you'll install the Python client library for InfluxDB and create an Amazon Timestream for InfluxDB instance on AWS. This gives you a managed time-series database that your fleet simulator will write to in the next step.

In this step, get ready to:
  • Install the InfluxDB Python client library.
  • Create a Timestream for InfluxDB database instance on AWS.
  • Retrieve your connection details and operator token.
Install the InfluxDB Python client library

Your Python scripts will use the influxdb-client package to connect to the database, write data points, and run queries. You need to install it before writing any code.

  • Open your terminal.
  • Install the InfluxDB client library by running this command:
pip install 'influxdb-client[ciso]'
  • Verify the installation succeeded by running:
python -c "from influxdb_client import InfluxDBClient, Point; print('Ready')"

You should see Ready printed in your terminal.

Not seeing Ready?

  • If you see a "ModuleNotFoundError", the install did not complete successfully. Run the pip install command again and check for error messages in the output.
  • If your system has both Python 2 and Python 3, try using pip3 instead of pip and python3 instead of python.
  • Help me troubleshoot my influxdb-client installation error.
Create a Timestream for InfluxDB instance

Amazon Timestream for InfluxDB is a managed database service that runs InfluxDB on AWS infrastructure. You create an instance through the AWS Console, and AWS handles the underlying servers, storage, and networking.

This step creates a billable AWS resource

The db.influx.medium instance you create costs money per hour while running. If you complete this project in 1-2 hours and delete the instance immediately after, total cost will be approximately $0.50-$1.50.

Make sure you follow the Before You Go section to delete your instance when finished.

First, you need to create a VPC security group that allows your computer to connect to the database on port 8086. Security groups cannot be changed after the instance is created, so you must set this up first.

  • Sign in to the AWS Management Console.
  • In the search bar at the top, type VPC and select VPC from the results.
  • In the left navigation pane, click Security Groups.
  • Click Create security group.
  • Enter swiftroute-influxdb-access for the Security group name.
  • Enter Allow InfluxDB access on port 8086 for the Description.
  • For VPC, select your default VPC.
  • In the Inbound rules section, click Add rule.
  • Set Type to Custom TCP.
  • Set Port range to 8086.
  • Set Source to My IP. This automatically fills in your current IP address.
  • Click Create security group.

Why create the security group first?

Once a Timestream for InfluxDB instance is created, you cannot change its associated security groups. You must have the security group ready before creating the instance so you can attach it during creation.

Now create the Timestream for InfluxDB database instance.

  • In the AWS Console search bar, type Timestream and select Amazon Timestream for InfluxDB.
  • In the left navigation pane, click InfluxDB Databases.
  • Click Create InfluxDB database.
  • For DB Instance Identifier, enter swiftroute-fleet.
  • For Initial username, enter admin.
  • For Initial organization name, enter swiftroute.
  • For Initial bucket name, enter fleet-telemetry.
  • For Password, enter a strong password and save it somewhere secure. You will need this to log in to the InfluxDB UI.
  • In the Instance configuration section, select db.influx.medium.
  • In the Storage configuration section, select Influx IO Included (3K) for the storage type.
  • In the Deployment section, select Single-AZ.
  • In the Connectivity configuration section, under Public access, select Publicly accessible.
  • For VPC Security Group (firewall), select the swiftroute-influxdb-access security group you just created.
  • Click Create InfluxDB database.

What is db.influx.medium?

This is the smallest available instance type for Timestream for InfluxDB. It provides 1 vCPU and 8 GiB of memory. AWS recommends it for development, testing, and low-volume workloads.

Your instance will take up to 20 minutes to become available. You can monitor its progress on the InfluxDB Databases page. Wait until the status changes to Available before continuing.

Use the wait time productively

While your instance is provisioning, review the project outline and think about how you would model delivery vehicle data as time-series points. Each vehicle reading has a timestamp, identifying tags, and measured values.

Instance stuck in Creating state?

  • The instance can take up to 20 minutes. If it still shows Creating after 20 minutes, check that your AWS region has available capacity for db.influx.medium instances.
  • Ensure your AWS account has no service quota limits blocking Timestream for InfluxDB instance creation.
  • Help me troubleshoot my Timestream for InfluxDB instance that won't reach Available status.
Retrieve your connection details and operator token

Your Python scripts need four pieces of information to connect: the instance endpoint URL, organization name, bucket name, and an API token. The operator token was generated automatically during instance setup and grants full access to all resources.

  • On the InfluxDB Databases page, click your swiftroute-fleet instance.
  • Copy the Endpoint URL shown on the instance details page. It will look like swiftroute-fleet-xxxxxxxx.timestream-influxdb.<region>.on.aws.
  • Open a new browser tab and navigate to your endpoint URL with port 8086 appended. For example: https://your-endpoint:8086.
  • Log in with the username admin and the password you set during creation.

You are now in the InfluxDB UI. Next, copy your operator token.

  • In the left navigation menu, click Load Data.
  • Click API Tokens.
  • Find the operator token in the list and click its name to reveal the token string.
  • Copy the token value and save it somewhere secure. You will need this in the next step.

What connection details do you need?

Save these four values. Your Python scripts will use them as environment variables to connect to the database:

  • Endpoint URL (with port 8086).
  • Organization name: swiftroute.
  • Bucket name: fleet-telemetry.
  • Operator token (the string you just copied).
  • Confirm that the fleet-telemetry bucket appears in the InfluxDB UI by clicking Load Data > Buckets in the left navigation. You should see fleet-telemetry listed with 0 bytes of data.

Can't access the InfluxDB UI in your browser?

  • Check that you included the port number (8086) in the URL.
  • Verify that your security group has an inbound rule for port 8086 with your current IP as the source. If your IP has changed since creating the rule, update the source to My IP again.
  • Confirm that the instance status is Available (not still Creating).
  • Help me troubleshoot connecting to my Timestream for InfluxDB UI in the browser.

Your database is live and ready for data. Next up, you'll build a fleet simulator that generates realistic delivery vehicle telemetry for all 10 SwiftRoute vans.

Build the Fleet Simulator

Your Timestream for InfluxDB instance is running and ready to receive data. Now you need something to generate that data.

SwiftRoute Logistics has 10 delivery vans on the road every morning. In this step, you'll build a Python simulator that recreates their telemetry: GPS coordinates, speed, fuel levels, and delivery statuses changing over a 4-hour route.

In this step, get ready to:
  • Create the project folder and define 10 vehicles with drivers and route zones.
  • Implement realistic delivery route patterns with traffic, fuel burn, and delivery stops.
  • Test the simulator and verify it generates ~4800 data points.
Create the project structure and simulator script

Every data pipeline starts with a source. Before you can ingest anything into your database, you need a script that produces structured telemetry data in the format InfluxDB expects: measurements, tags, fields, and timestamps.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir swiftroute-fleet
cd swiftroute-fleet
  • Confirm the folder was created by running:
ls

The output should be empty since the folder is brand new.

  • Open your code editor.
  • Create a new file called fleet_simulator.py inside the swiftroute-fleet folder.
  • Add the vehicle fleet definitions and the orchestrator function by pasting this code into the file:
import random
from datetime import datetime, timedelta, timezone


VEHICLES = [
    {"vehicle_id": "VAN-001", "driver": "Carlos", "route_zone": "north"},
    {"vehicle_id": "VAN-002", "driver": "Maria", "route_zone": "south"},
    {"vehicle_id": "VAN-003", "driver": "James", "route_zone": "east"},
    {"vehicle_id": "VAN-004", "driver": "Priya", "route_zone": "west"},
    {"vehicle_id": "VAN-005", "driver": "Ahmed", "route_zone": "north"},
    {"vehicle_id": "VAN-006", "driver": "Sofia", "route_zone": "south"},
    {"vehicle_id": "VAN-007", "driver": "Liam", "route_zone": "east"},
    {"vehicle_id": "VAN-008", "driver": "Yuki", "route_zone": "west"},
    {"vehicle_id": "VAN-009", "driver": "Omar", "route_zone": "north"},
    {"vehicle_id": "VAN-010", "driver": "Elena", "route_zone": "south"},
]

WAREHOUSE_LAT = 40.7128
WAREHOUSE_LNG = -74.0060

INTERVAL_SECONDS = 30
DURATION_HOURS = 4


def generate_fleet_telemetry():
    """Generate 4 hours of telemetry data for all vehicles."""
    all_points = []

    start_time = datetime.now(tz=timezone.utc) - timedelta(hours=DURATION_HOURS)

    for vehicle in VEHICLES:
        points = generate_vehicle_route(vehicle, start_time)
        all_points.extend(points)

    all_points.sort(key=lambda p: p["timestamp"])
    return all_points

What does this code do?

This sets up the foundation of the simulator.

  • VEHICLES defines 10 delivery vans. Each van has a unique ID, a driver name, and a route zone (north, south, east, or west). These become tags in InfluxDB, which means you can filter and group queries by them later.
  • WAREHOUSE_LAT and WAREHOUSE_LNG set the starting GPS coordinates for all vans (Manhattan, New York).
  • generate_fleet_telemetry() is the entry point. It loops through all 10 vehicles, generates each route, then sorts all data points by timestamp so they interleave chronologically.
  • Save fleet_simulator.py.

You won't see any output yet. The next chunk adds the route generation logic that this function calls.

Implement realistic delivery route patterns

A realistic simulator needs to mirror what actually happens on a delivery route. Vans leave the warehouse at slightly different times, hit traffic during rush hour, stop for deliveries, and burn fuel at different rates depending on whether they're moving or idling.

  • In fleet_simulator.py, add the route generation function below the generate_fleet_telemetry() function:
def generate_vehicle_route(vehicle, start_time):
    """Generate a realistic delivery route for one vehicle."""
    points = []
    total_intervals = (DURATION_HOURS * 3600) // INTERVAL_SECONDS

    lat = WAREHOUSE_LAT + random.uniform(-0.01, 0.01)
    lng = WAREHOUSE_LNG + random.uniform(-0.01, 0.01)
    fuel_level = random.uniform(85.0, 100.0)
    delivery_status = "idle"
    speed = 0.0
    engine_on = False

    departure_offset = random.randint(0, 20)
    delivery_stops = sorted(random.sample(range(30, total_intervals - 30), random.randint(6, 10)))
    delivery_durations = {stop: random.randint(6, 16) for stop in delivery_stops}

    current_delivery_end = 0

    for i in range(total_intervals):
        timestamp = start_time + timedelta(seconds=i * INTERVAL_SECONDS)
        hour_of_day = (start_time.hour + (i * INTERVAL_SECONDS) / 3600) % 24

        if i < departure_offset:
            speed = 0.0
            engine_on = False
            delivery_status = "idle"
        elif i in delivery_stops:
            speed = 0.0
            engine_on = True
            delivery_status = "delivering"
            current_delivery_end = i + delivery_durations[i]
        elif i < current_delivery_end:
            speed = 0.0
            engine_on = True
            delivery_status = "delivering"
        elif i >= total_intervals - 30:
            engine_on = True
            delivery_status = "returning"
            speed = random.uniform(15.0, 40.0)
        else:
            engine_on = True
            delivery_status = "in_transit"

            if 8.0 <= hour_of_day <= 9.0 or 17.0 <= hour_of_day <= 18.0:
                speed = random.uniform(5.0, 25.0)
            else:
                speed = random.uniform(20.0, 55.0)

            if random.random() < 0.05:
                speed = 0.0
                delivery_status = "idle"
                engine_on = True

        if engine_on and speed > 0:
            lat += random.uniform(-0.002, 0.002)
            lng += random.uniform(-0.002, 0.002)
            fuel_level -= random.uniform(0.02, 0.08)
        elif engine_on and speed == 0:
            fuel_level -= random.uniform(0.005, 0.015)

        fuel_level = max(fuel_level, 5.0)

        points.append({
            "timestamp": timestamp,
            "measurement": "vehicle_telemetry",
            "tags": {
                "vehicle_id": vehicle["vehicle_id"],
                "driver": vehicle["driver"],
                "route_zone": vehicle["route_zone"],
            },
            "fields": {
                "latitude": round(lat, 6),
                "longitude": round(lng, 6),
                "speed_kmh": round(speed, 1),
                "fuel_level_pct": round(fuel_level, 2),
                "engine_on": engine_on,
                "delivery_status": delivery_status,
            },
        })

    return points

What's happening in this function?

This function simulates one vehicle's entire 4-hour morning route. Each loop iteration represents a 30-second interval, producing one telemetry data point.

  • departure_offset randomizes when each van leaves the warehouse (within 10 minutes of each other).
  • delivery_stops picks 6-10 random intervals where the van stops to deliver. Each stop lasts 3-8 minutes.
  • Traffic patterns slow speeds during rush hours (0800-0900 and 1700-1800).
  • Fuel decreases faster when moving (0.02-0.08% per interval) and slower when idling (0.005-0.015%).
  • Each data point is structured with tags (vehicle_id, driver, route_zone) and fields (latitude, longitude, speed, fuel, engine_on, delivery_status). This is the exact structure InfluxDB expects.

Why tags vs fields?

In InfluxDB, tags are indexed metadata you filter and group by (like vehicle_id or route_zone). Fields are the actual measured values (like speed or fuel level). Tags are fast to query but expensive to store many unique values. Fields store the numbers but aren't indexed.

  • Save fleet_simulator.py.

You won't see output yet. The next section adds a test block that runs the simulator and prints results.

Test the simulator output locally

Before sending data to your database, you should verify the simulator produces correctly structured output. A quick test run confirms the data looks realistic.

  • Add the test block at the bottom of fleet_simulator.py, below the generate_vehicle_route() function:
if __name__ == "__main__":
    data = generate_fleet_telemetry()
    print(f"Generated {len(data)} telemetry points for {len(VEHICLES)} vehicles")
    print(f"\nSample point (first):")
    sample = data[0]
    print(f"  Time: {sample['timestamp']}")
    print(f"  Vehicle: {sample['tags']['vehicle_id']} ({sample['tags']['driver']})")
    print(f"  Zone: {sample['tags']['route_zone']}")
    print(f"  Speed: {sample['fields']['speed_kmh']} km/h")
    print(f"  Fuel: {sample['fields']['fuel_level_pct']}%")
    print(f"  Status: {sample['fields']['delivery_status']}")

What does this code do?

The if __name__ == "__main__": block only runs when you execute the file directly. It generates the full dataset and prints a summary with one sample point so you can verify the structure is correct.

  • Save fleet_simulator.py.
  • Run the simulator by executing this command in your terminal:
python fleet_simulator.py

You should see output showing approximately 4800 generated telemetry points, followed by a sample data point with a timestamp, vehicle ID, driver name, zone, speed, fuel percentage, and delivery status.

Seeing a ModuleNotFoundError or syntax error?

  • Make sure your terminal is inside the swiftroute-fleet folder. Run pwd to check.
  • Check that you saved the file after adding all three sections (constants, generate_fleet_telemetry, and generate_vehicle_route).
  • If you see an indentation error, make sure the for loop inside generate_vehicle_route uses consistent indentation (4 spaces per level).
  • My simulator isn't producing the expected output. Help me debug.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete fleet_simulator.py file for comparison:

import random
from datetime import datetime, timedelta, timezone


VEHICLES = [
    {"vehicle_id": "VAN-001", "driver": "Carlos", "route_zone": "north"},
    {"vehicle_id": "VAN-002", "driver": "Maria", "route_zone": "south"},
    {"vehicle_id": "VAN-003", "driver": "James", "route_zone": "east"},
    {"vehicle_id": "VAN-004", "driver": "Priya", "route_zone": "west"},
    {"vehicle_id": "VAN-005", "driver": "Ahmed", "route_zone": "north"},
    {"vehicle_id": "VAN-006", "driver": "Sofia", "route_zone": "south"},
    {"vehicle_id": "VAN-007", "driver": "Liam", "route_zone": "east"},
    {"vehicle_id": "VAN-008", "driver": "Yuki", "route_zone": "west"},
    {"vehicle_id": "VAN-009", "driver": "Omar", "route_zone": "north"},
    {"vehicle_id": "VAN-010", "driver": "Elena", "route_zone": "south"},
]

WAREHOUSE_LAT = 40.7128
WAREHOUSE_LNG = -74.0060

INTERVAL_SECONDS = 30
DURATION_HOURS = 4


def generate_fleet_telemetry():
    """Generate 4 hours of telemetry data for all vehicles."""
    all_points = []

    start_time = datetime.now(tz=timezone.utc) - timedelta(hours=DURATION_HOURS)

    for vehicle in VEHICLES:
        points = generate_vehicle_route(vehicle, start_time)
        all_points.extend(points)

    all_points.sort(key=lambda p: p["timestamp"])
    return all_points


def generate_vehicle_route(vehicle, start_time):
    """Generate a realistic delivery route for one vehicle."""
    points = []
    total_intervals = (DURATION_HOURS * 3600) // INTERVAL_SECONDS

    lat = WAREHOUSE_LAT + random.uniform(-0.01, 0.01)
    lng = WAREHOUSE_LNG + random.uniform(-0.01, 0.01)
    fuel_level = random.uniform(85.0, 100.0)
    delivery_status = "idle"
    speed = 0.0
    engine_on = False

    departure_offset = random.randint(0, 20)
    delivery_stops = sorted(random.sample(range(30, total_intervals - 30), random.randint(6, 10)))
    delivery_durations = {stop: random.randint(6, 16) for stop in delivery_stops}

    current_delivery_end = 0

    for i in range(total_intervals):
        timestamp = start_time + timedelta(seconds=i * INTERVAL_SECONDS)
        hour_of_day = (start_time.hour + (i * INTERVAL_SECONDS) / 3600) % 24

        if i < departure_offset:
            speed = 0.0
            engine_on = False
            delivery_status = "idle"
        elif i in delivery_stops:
            speed = 0.0
            engine_on = True
            delivery_status = "delivering"
            current_delivery_end = i + delivery_durations[i]
        elif i < current_delivery_end:
            speed = 0.0
            engine_on = True
            delivery_status = "delivering"
        elif i >= total_intervals - 30:
            engine_on = True
            delivery_status = "returning"
            speed = random.uniform(15.0, 40.0)
        else:
            engine_on = True
            delivery_status = "in_transit"

            if 8.0 <= hour_of_day <= 9.0 or 17.0 <= hour_of_day <= 18.0:
                speed = random.uniform(5.0, 25.0)
            else:
                speed = random.uniform(20.0, 55.0)

            if random.random() < 0.05:
                speed = 0.0
                delivery_status = "idle"
                engine_on = True

        if engine_on and speed > 0:
            lat += random.uniform(-0.002, 0.002)
            lng += random.uniform(-0.002, 0.002)
            fuel_level -= random.uniform(0.02, 0.08)
        elif engine_on and speed == 0:
            fuel_level -= random.uniform(0.005, 0.015)

        fuel_level = max(fuel_level, 5.0)

        points.append({
            "timestamp": timestamp,
            "measurement": "vehicle_telemetry",
            "tags": {
                "vehicle_id": vehicle["vehicle_id"],
                "driver": vehicle["driver"],
                "route_zone": vehicle["route_zone"],
            },
            "fields": {
                "latitude": round(lat, 6),
                "longitude": round(lng, 6),
                "speed_kmh": round(speed, 1),
                "fuel_level_pct": round(fuel_level, 2),
                "engine_on": engine_on,
                "delivery_status": delivery_status,
            },
        })

    return points


if __name__ == "__main__":
    data = generate_fleet_telemetry()
    print(f"Generated {len(data)} telemetry points for {len(VEHICLES)} vehicles")
    print(f"\nSample point (first):")
    sample = data[0]
    print(f"  Time: {sample['timestamp']}")
    print(f"  Vehicle: {sample['tags']['vehicle_id']} ({sample['tags']['driver']})")
    print(f"  Zone: {sample['tags']['route_zone']}")
    print(f"  Speed: {sample['fields']['speed_kmh']} km/h")
    print(f"  Fuel: {sample['fields']['fuel_level_pct']}%")
    print(f"  Status: {sample['fields']['delivery_status']}")

Your fleet simulator is generating realistic telemetry data. Next up, you'll send all 4800 data points into your Timestream for InfluxDB instance so you can start running queries against them.

Ingest Fleet Telemetry into Timestream

Your simulator generates realistic delivery data, but it only lives in memory. To answer business questions about SLA breaches or fuel efficiency, you need that data stored in a time-series database where it can be queried across time windows.

In this step, you will build an ingestion script that connects to your Timestream for InfluxDB instance, converts each telemetry record into an InfluxDB Point, and writes the entire dataset in batches.

In this step, get ready to:
  • Set environment variables and create the ingestion script with connection configuration.
  • Write batch ingestion logic that converts simulator output into InfluxDB Points.
  • Run the ingestion and verify ~4800 data points landed in InfluxDB.
Set your connection credentials

Your ingestion script needs four pieces of information to reach your Timestream for InfluxDB instance: the endpoint URL, your operator token, the organization name, and the bucket name. You will store these as environment variables so the script can read them at runtime without hardcoding sensitive values.

  • Set the four connection environment variables by running this command in your terminal:
export INFLUX_URL=[[INFLUX_URL="https://your-instance-endpoint:8086"]]
export INFLUX_TOKEN=your-operator-token-here
export INFLUX_ORG=swiftroute
export INFLUX_BUCKET=fleet-telemetry
  • Replace the INFLUX_URL value with your Timestream for InfluxDB endpoint URL (the one you used to access the InfluxDB UI, including :8086).
  • Replace your-operator-token-here with the operator token you copied from the InfluxDB UI in the previous step.

Why environment variables?

Storing credentials in environment variables keeps them out of your source code. The script reads them at runtime with fallback defaults, so you never risk accidentally committing a token to version control.

Now create the ingestion script file.

  • Create a new file called ingest_data.py in your swiftroute-fleet folder and add the imports and configuration:
import os
import sys
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
from fleet_simulator import generate_fleet_telemetry


# Read connection details from environment variables
INFLUX_URL = os.getenv("INFLUX_URL", "http://localhost:8086")
INFLUX_TOKEN = os.getenv("INFLUX_TOKEN", "your-token-here")
INFLUX_ORG = os.getenv("INFLUX_ORG", "swiftroute")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "fleet-telemetry")

# Number of points to send per write call
BATCH_SIZE = 500

What does this code do?

The imports bring in the InfluxDB Python client classes you need: InfluxDBClient for the connection, Point for building data records, WritePrecision for timestamp resolution, and SYNCHRONOUS for blocking writes.

The os.getenv() calls read your environment variables with sensible fallbacks. BATCH_SIZE controls how many points get sent per network request, balancing speed against memory usage.

  • Save ingest_data.py.

Import error when you save?

If your editor flags from fleet_simulator import generate_fleet_telemetry as unresolved, make sure fleet_simulator.py is in the same folder as ingest_data.py. Both files must live inside swiftroute-fleet.

help me fix my import error for fleet_simulator

Write the batch ingestion logic

Each telemetry record from the simulator is a Python dictionary. InfluxDB expects data as Point objects with explicit tags, fields, and timestamps. You need a function to handle that conversion, plus a main function that orchestrates generating data, writing it in batches, and reporting progress.

  • Add the create_point() function below your configuration constants in ingest_data.py:
def create_point(record):
    """Convert a telemetry record to an InfluxDB Point."""
    point = (
        Point(record["measurement"])
        .tag("vehicle_id", record["tags"]["vehicle_id"])
        .tag("driver", record["tags"]["driver"])
        .tag("route_zone", record["tags"]["route_zone"])
        .field("latitude", record["fields"]["latitude"])
        .field("longitude", record["fields"]["longitude"])
        .field("speed_kmh", record["fields"]["speed_kmh"])
        .field("fuel_level_pct", record["fields"]["fuel_level_pct"])
        .field("engine_on", record["fields"]["engine_on"])
        .field("delivery_status", record["fields"]["delivery_status"])
        .time(record["timestamp"], WritePrecision.S)
    )
    return point

What does this code do?

This function maps each simulator record into an InfluxDB Point. Tags (vehicle_id, driver, route_zone) are indexed metadata you can filter and group by efficiently. Fields (latitude, speed_kmh, etc.) are the measured values you aggregate and compute on.

The WritePrecision.S tells InfluxDB that timestamps are in seconds. This matches the 30-second intervals your simulator generates.

  • Save ingest_data.py.

Now add the main ingestion function that ties everything together.

  • Add the ingest_fleet_data() function below create_point() in ingest_data.py:
def ingest_fleet_data():
    """Generate and ingest fleet telemetry into Timestream for InfluxDB."""
    print("Generating fleet telemetry data...")
    telemetry_data = generate_fleet_telemetry()
    print(f"Generated {len(telemetry_data)} data points")

    print(f"\nConnecting to InfluxDB at {INFLUX_URL}...")
    with InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG) as client:
        write_api = client.write_api(write_options=SYNCHRONOUS)

        print(f"Writing data to bucket '{INFLUX_BUCKET}'...")
        total = len(telemetry_data)
        for i in range(0, total, BATCH_SIZE):
            batch = telemetry_data[i:i + BATCH_SIZE]
            points = [create_point(record) for record in batch]
            write_api.write(bucket=INFLUX_BUCKET, org=INFLUX_ORG, record=points)
            written = min(i + BATCH_SIZE, total)
            print(f"  Progress: {written}/{total} points written")

        print(f"\nIngestion complete. {total} points written to '{INFLUX_BUCKET}'.")

    verify_data()

What does this code do?

This function runs the full ingestion pipeline. It generates the telemetry data by calling your simulator, opens a synchronous connection to InfluxDB, then iterates through the data in batches of 500 points.

Each batch converts records to Points using create_point() and writes them in a single network call. The progress counter lets you see ingestion happening in real time. After all batches complete, it calls verify_data() (which you will add next) to confirm the data landed.

  • Save ingest_data.py.

Why tags vs fields?

In a time-series database, tags are indexed and fields are not. This means filtering by vehicle_id or grouping by route_zone is fast because InfluxDB can look them up in an index. Fields like speed_kmh and fuel_level_pct store the actual measured values you want to aggregate (averages, sums, max).

A good rule: if you will filter or group by it, make it a tag. If you will compute on it (sum, mean, count), make it a field.

Run the ingestion and verify your data

The last piece is a verification function that runs a Flux query to count how many records actually made it into the database. This confirms your ingestion worked end-to-end before you move on to business queries.

  • Add the verify_data() function and the __main__ block at the bottom of ingest_data.py:
def verify_data():
    """Run a quick count query to verify data landed."""
    with InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG) as client:
        query_api = client.query_api()
        query = f'''
        from(bucket: "{INFLUX_BUCKET}")
          |> range(start: -5h)
          |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
          |> filter(fn: (r) => r._field == "speed_kmh")
          |> count()
          |> group()
          |> sum(column: "_value")
        '''
        tables = query_api.query(query, org=INFLUX_ORG)
        for table in tables:
            for record in table.records:
                print(f"\nVerification: {record.get_value()} speed_kmh records found in InfluxDB")


if __name__ == "__main__":
    ingest_fleet_data()

What does this Flux query do?

This Flux query reads from the fleet-telemetry bucket, filters for the vehicle_telemetry measurement and the speed_kmh field, counts records per series, then sums those counts into a single total. You should see approximately 4800 records (10 vehicles x 480 intervals each).

  • Save ingest_data.py.
  • Run the ingestion script by executing this command:
python ingest_data.py
  • Watch the progress output. You should see batches writing (500 points at a time) followed by a verification line showing approximately 4800 records.

Seeing an unauthorized error?

If you see a 401 Unauthorized error, your environment variables are not reaching the script correctly. Double-check these common causes:

  • Make sure you ran the export commands in the same terminal session where you run python ingest_data.py. Opening a new terminal tab resets environment variables.
  • Verify your INFLUX_TOKEN is the actual token string from the InfluxDB UI (Load Data > API Tokens), not the token name.
  • Confirm your INFLUX_URL includes the port (:8086) and uses the correct protocol (https://).

Help me debug my InfluxDB unauthorized error

Now verify the data visually in the InfluxDB UI.

  • Return to the InfluxDB UI in your browser.
  • Click **Data Explorer** in the left sidebar.
  • Select the fleet-telemetry bucket.
  • Under the vehicle_telemetry measurement, select speed_kmh as the field.
  • Filter by a single vehicle_id tag (for example, VAN-001) to see one vehicle's speed over time.
  • Click **Submit** to run the query. You should see a line chart showing speed fluctuations over the 4-hour window.

✔️ Awesome, I've got everything!

Great. Double check you have saved ingest_data.py and that your terminal output shows approximately 4800 records verified.

ⓧ I'd like to double check the full code

Here is the complete ingest_data.py file for comparison:

import os
import sys
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
from fleet_simulator import generate_fleet_telemetry


INFLUX_URL = os.getenv("INFLUX_URL", "http://localhost:8086")
INFLUX_TOKEN = os.getenv("INFLUX_TOKEN", "your-token-here")
INFLUX_ORG = os.getenv("INFLUX_ORG", "swiftroute")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "fleet-telemetry")

BATCH_SIZE = 500


def create_point(record):
    """Convert a telemetry record to an InfluxDB Point."""
    point = (
        Point(record["measurement"])
        .tag("vehicle_id", record["tags"]["vehicle_id"])
        .tag("driver", record["tags"]["driver"])
        .tag("route_zone", record["tags"]["route_zone"])
        .field("latitude", record["fields"]["latitude"])
        .field("longitude", record["fields"]["longitude"])
        .field("speed_kmh", record["fields"]["speed_kmh"])
        .field("fuel_level_pct", record["fields"]["fuel_level_pct"])
        .field("engine_on", record["fields"]["engine_on"])
        .field("delivery_status", record["fields"]["delivery_status"])
        .time(record["timestamp"], WritePrecision.S)
    )
    return point


def ingest_fleet_data():
    """Generate and ingest fleet telemetry into Timestream for InfluxDB."""
    print("Generating fleet telemetry data...")
    telemetry_data = generate_fleet_telemetry()
    print(f"Generated {len(telemetry_data)} data points")

    print(f"\nConnecting to InfluxDB at {INFLUX_URL}...")
    with InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG) as client:
        write_api = client.write_api(write_options=SYNCHRONOUS)

        print(f"Writing data to bucket '{INFLUX_BUCKET}'...")
        total = len(telemetry_data)
        for i in range(0, total, BATCH_SIZE):
            batch = telemetry_data[i:i + BATCH_SIZE]
            points = [create_point(record) for record in batch]
            write_api.write(bucket=INFLUX_BUCKET, org=INFLUX_ORG, record=points)
            written = min(i + BATCH_SIZE, total)
            print(f"  Progress: {written}/{total} points written")

        print(f"\nIngestion complete. {total} points written to '{INFLUX_BUCKET}'.")

    verify_data()


def verify_data():
    """Run a quick count query to verify data landed."""
    with InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG) as client:
        query_api = client.query_api()
        query = f'''
        from(bucket: "{INFLUX_BUCKET}")
          |> range(start: -5h)
          |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
          |> filter(fn: (r) => r._field == "speed_kmh")
          |> count()
          |> group()
          |> sum(column: "_value")
        '''
        tables = query_api.query(query, org=INFLUX_ORG)
        for table in tables:
            for record in table.records:
                print(f"\nVerification: {record.get_value()} speed_kmh records found in InfluxDB")


if __name__ == "__main__":
    ingest_fleet_data()

Your fleet telemetry is now stored in a time-series database, ready to be queried. Next up, you will write Flux queries that answer real business questions for SwiftRoute: which drivers are at risk of missing SLA windows, who is burning the most fuel, and where idle time is costing money.

Query for Supply Chain Business Insights

Your fleet telemetry is now sitting in Timestream for InfluxDB, ready to be analyzed.

SwiftRoute's operations team needs answers to three urgent business questions: which vehicles are at risk of breaching their 2-hour delivery SLA, which vans are burning the most fuel, and which drivers are wasting fuel by idling. In this step, you will write Flux queries that turn raw time-series data into actionable supply chain reports.

In this step, get ready to:
  • Create a query script that connects to your InfluxDB instance.
  • Write Flux queries for SLA breach detection and fuel efficiency ranking.
  • Build an idle time analysis report and run all three business queries.
Create the query script with connection logic

Each query function will connect to InfluxDB independently using a shared helper. This keeps the code modular so you can run individual reports later.

  • Create a new file called query_insights.py inside your swiftroute-fleet folder.
  • Add the connection configuration and client helper by pasting this code:
import os
from influxdb_client import InfluxDBClient


# Connection settings from environment variables
INFLUX_URL = os.getenv("INFLUX_URL", "http://localhost:8086")
INFLUX_TOKEN = os.getenv("INFLUX_TOKEN", "your-token-here")
INFLUX_ORG = os.getenv("INFLUX_ORG", "swiftroute")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "fleet-telemetry")


def get_client():
    """Create and return an InfluxDB client."""
    return InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)

What does this code do?

The connection constants pull from the same environment variables you set for the ingestion script.

The get_client() helper creates a reusable InfluxDB client. Each query function calls this independently so the connection is cleanly opened and closed per report.

  • Save query_insights.py.
Build the SLA breach detection and fuel efficiency queries

SwiftRoute guarantees deliveries within a 2-hour window. If a vehicle spends too long between deliveries, it risks breaching the SLA.

  • Add the SLA breach risk function below get_client() in query_insights.py:
def query_sla_breach_risk():
    """Find vehicles with long gaps between deliveries (approaching 2-hour SLA)."""
    print("\n" + "=" * 60)
    print("REPORT 1: SLA Breach Risk - Vehicles Approaching 2-Hour Window")
    print("=" * 60)

    query = f'''
    from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -5h)
      |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
      |> filter(fn: (r) => r._field == "delivery_status")
      |> filter(fn: (r) => r._value == "in_transit" or r._value == "idle")
      |> group(columns: ["vehicle_id", "driver"])
      |> elapsed(unit: 1m)
      |> filter(fn: (r) => r.elapsed >= 60)
      |> max(column: "elapsed")
      |> group()
      |> sort(columns: ["elapsed"], desc: true)
      |> limit(n: 10)
    '''

    with get_client() as client:
        query_api = client.query_api()
        tables = query_api.query(query, org=INFLUX_ORG)

        print(f"\n{'Vehicle':<12} {'Driver':<10} {'Max Gap (min)':<15} {'Risk Level'}")
        print("-" * 50)
        for table in tables:
            for record in table.records:
                vehicle = record.values.get("vehicle_id", "N/A")
                driver = record.values.get("driver", "N/A")
                gap_min = record.get_value()
                risk = "HIGH" if gap_min >= 90 else "MEDIUM" if gap_min >= 60 else "LOW"
                print(f"{vehicle:<12} {driver:<10} {gap_min:<15} {risk}")

How does the SLA breach query work?

  • The query filters for data points where the vehicle status is in_transit or idle. These are periods between deliveries.
  • The Flux elapsed() function calculates the time gap between consecutive records in minutes.
  • Gaps of 60+ minutes are flagged. 90+ minutes is HIGH risk (approaching the 2-hour SLA window).
  • Results are sorted so the worst offenders appear first.

Next, the fuel efficiency query ranks vehicles by how much fuel they consumed over the monitoring period.

  • Add the fuel efficiency function directly below query_sla_breach_risk():
def query_fuel_efficiency():
    """Rank vehicles by fuel consumption over the monitoring period."""
    print("\n" + "=" * 60)
    print("REPORT 2: Fuel Efficiency Ranking")
    print("=" * 60)

    query = f'''
    from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -5h)
      |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
      |> filter(fn: (r) => r._field == "fuel_level_pct")
      |> group(columns: ["vehicle_id", "driver", "route_zone"])
      |> reduce(
          fn: (r, accumulator) => ({{
            first: if accumulator.count == 0 then r._value else accumulator.first,
            last: r._value,
            count: accumulator.count + 1
          }}),
          identity: {{first: 0.0, last: 0.0, count: 0}}
      )
      |> map(fn: (r) => ({{r with fuel_consumed: r.first - r.last}}))
      |> group()
      |> sort(columns: ["fuel_consumed"], desc: true)
    '''

    with get_client() as client:
        query_api = client.query_api()
        tables = query_api.query(query, org=INFLUX_ORG)

        print(f"\n{'Rank':<6} {'Vehicle':<12} {'Driver':<10} {'Zone':<8} {'Fuel Used (%)'}")
        print("-" * 55)
        rank = 1
        for table in tables:
            for record in table.records:
                vehicle = record.values.get("vehicle_id", "N/A")
                driver = record.values.get("driver", "N/A")
                zone = record.values.get("route_zone", "N/A")
                fuel_used = record.values.get("fuel_consumed", 0)
                print(f"{rank:<6} {vehicle:<12} {driver:<10} {zone:<8} {fuel_used:.1f}%")
                rank += 1

How does the fuel efficiency query work?

  • The Flux reduce() function captures the first and last fuel reading for each vehicle.
  • The map() function calculates fuel_consumed as the difference between the first and last readings.
  • Vehicles are ranked with the highest fuel consumers at the top. These are the vans that may need maintenance or route optimization.

Why use reduce() instead of first() and last()?

Flux's reduce() processes records sequentially in a single pass, capturing both the first value and tracking the latest value as it goes. This approach handles the calculation in one query pipeline without needing to join two separate result sets.

  • Save query_insights.py.

Seeing a syntax error in the Flux query?

The double curly braces {{ and }} are required because the query string is an f-string in Python. Single braces would be interpreted as Python format placeholders.

help me debug a Flux query syntax error in my Python f-string

Build the idle time analysis query and run the reports

The final query identifies drivers who burn fuel while stationary. A vehicle with the engine running and zero speed is wasting fuel without making progress on deliveries.

  • Add the idle time function below query_fuel_efficiency() in query_insights.py:
def query_idle_time():
    """Calculate idle time (engine on, speed 0, not delivering) per driver."""
    print("\n" + "=" * 60)
    print("REPORT 3: Idle Time Analysis - Fuel Burn Without Movement")
    print("=" * 60)

    query = f'''
    from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -5h)
      |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
      |> filter(fn: (r) => r._field == "speed_kmh")
      |> filter(fn: (r) => r._value == 0.0)
      |> group(columns: ["vehicle_id", "driver", "route_zone"])
      |> count()
      |> map(fn: (r) => ({{r with idle_minutes: r._value / 2}}))
      |> group()
      |> sort(columns: ["idle_minutes"], desc: true)
    '''

    with get_client() as client:
        query_api = client.query_api()
        tables = query_api.query(query, org=INFLUX_ORG)

        print(f"\n{'Vehicle':<12} {'Driver':<10} {'Zone':<8} {'Idle Time (min)':<18} {'Status'}")
        print("-" * 60)
        for table in tables:
            for record in table.records:
                vehicle = record.values.get("vehicle_id", "N/A")
                driver = record.values.get("driver", "N/A")
                zone = record.values.get("route_zone", "N/A")
                idle_min = record.values.get("idle_minutes", 0)
                status = "EXCESSIVE" if idle_min > 40 else "NORMAL"
                print(f"{vehicle:<12} {driver:<10} {zone:<8} {idle_min:<18.0f} {status}")

    print("\n>> RECOMMENDATION: Drivers with EXCESSIVE idle time may need")
    print("   route optimization or dispatch prioritization.")

How does the idle time query work?

  • The query filters for records where speed_kmh is exactly 0. This captures all stationary moments (delivery stops, traffic, and pure idle time).
  • It counts these zero-speed records per vehicle, then divides by 2 to convert to minutes (since data points arrive every 30 seconds).
  • Drivers with more than 40 minutes of idle time are flagged as EXCESSIVE.

Now add the main() function that ties all three reports together.

  • Add the main function and entry point at the bottom of query_insights.py:
def main():
    print("SwiftRoute Logistics - Fleet Analytics Dashboard")
    print("Querying Timestream for InfluxDB...")

    query_sla_breach_risk()
    query_fuel_efficiency()
    query_idle_time()

    print("\n" + "=" * 60)
    print("All reports generated successfully.")
    print("=" * 60)


if __name__ == "__main__":
    main()

What does this code do?

The main() function calls all three query functions in sequence, producing a complete analytics dashboard in the terminal. Each report runs its own InfluxDB connection and prints formatted results.

  • Save query_insights.py.
  • Run the query script by executing this command in your terminal:
python query_insights.py
  • You should see three formatted reports printed to the terminal: SLA Breach Risk showing vehicles with HIGH/MEDIUM risk levels, Fuel Efficiency Ranking listing all 10 vehicles by fuel consumption, and Idle Time Analysis flagging drivers with EXCESSIVE idle time.

Seeing an empty report or connection error?

  • Confirm your environment variables are still set by running echo $INFLUX_URL in your terminal. If empty, re-export them.
  • Check that your Timestream for InfluxDB instance is still in the Available state in the AWS Console.
  • If reports are empty but the script connects successfully, verify you ran ingest_data.py within the last 5 hours. The queries use a range(start: -5h) filter.

help me debug why my Flux query returns empty results

✔️ Awesome, I've got everything!

Great. Double check you have saved query_insights.py and that all three reports printed successfully.

ⓧ I'd like to double check the full code

import os
from influxdb_client import InfluxDBClient


INFLUX_URL = os.getenv("INFLUX_URL", "http://localhost:8086")
INFLUX_TOKEN = os.getenv("INFLUX_TOKEN", "your-token-here")
INFLUX_ORG = os.getenv("INFLUX_ORG", "swiftroute")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "fleet-telemetry")


def get_client():
    """Create and return an InfluxDB client."""
    return InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)


def query_sla_breach_risk():
    """Find vehicles with long gaps between deliveries (approaching 2-hour SLA)."""
    print("\n" + "=" * 60)
    print("REPORT 1: SLA Breach Risk - Vehicles Approaching 2-Hour Window")
    print("=" * 60)

    query = f'''
    from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -5h)
      |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
      |> filter(fn: (r) => r._field == "delivery_status")
      |> filter(fn: (r) => r._value == "in_transit" or r._value == "idle")
      |> group(columns: ["vehicle_id", "driver"])
      |> elapsed(unit: 1m)
      |> filter(fn: (r) => r.elapsed >= 60)
      |> max(column: "elapsed")
      |> group()
      |> sort(columns: ["elapsed"], desc: true)
      |> limit(n: 10)
    '''

    with get_client() as client:
        query_api = client.query_api()
        tables = query_api.query(query, org=INFLUX_ORG)

        print(f"\n{'Vehicle':<12} {'Driver':<10} {'Max Gap (min)':<15} {'Risk Level'}")
        print("-" * 50)
        for table in tables:
            for record in table.records:
                vehicle = record.values.get("vehicle_id", "N/A")
                driver = record.values.get("driver", "N/A")
                gap_min = record.get_value()
                risk = "HIGH" if gap_min >= 90 else "MEDIUM" if gap_min >= 60 else "LOW"
                print(f"{vehicle:<12} {driver:<10} {gap_min:<15} {risk}")


def query_fuel_efficiency():
    """Rank vehicles by fuel consumption over the monitoring period."""
    print("\n" + "=" * 60)
    print("REPORT 2: Fuel Efficiency Ranking")
    print("=" * 60)

    query = f'''
    from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -5h)
      |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
      |> filter(fn: (r) => r._field == "fuel_level_pct")
      |> group(columns: ["vehicle_id", "driver", "route_zone"])
      |> reduce(
          fn: (r, accumulator) => ({{
            first: if accumulator.count == 0 then r._value else accumulator.first,
            last: r._value,
            count: accumulator.count + 1
          }}),
          identity: {{first: 0.0, last: 0.0, count: 0}}
      )
      |> map(fn: (r) => ({{r with fuel_consumed: r.first - r.last}}))
      |> group()
      |> sort(columns: ["fuel_consumed"], desc: true)
    '''

    with get_client() as client:
        query_api = client.query_api()
        tables = query_api.query(query, org=INFLUX_ORG)

        print(f"\n{'Rank':<6} {'Vehicle':<12} {'Driver':<10} {'Zone':<8} {'Fuel Used (%)'}")
        print("-" * 55)
        rank = 1
        for table in tables:
            for record in table.records:
                vehicle = record.values.get("vehicle_id", "N/A")
                driver = record.values.get("driver", "N/A")
                zone = record.values.get("route_zone", "N/A")
                fuel_used = record.values.get("fuel_consumed", 0)
                print(f"{rank:<6} {vehicle:<12} {driver:<10} {zone:<8} {fuel_used:.1f}%")
                rank += 1


def query_idle_time():
    """Calculate idle time (engine on, speed 0, not delivering) per driver."""
    print("\n" + "=" * 60)
    print("REPORT 3: Idle Time Analysis - Fuel Burn Without Movement")
    print("=" * 60)

    query = f'''
    from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -5h)
      |> filter(fn: (r) => r._measurement == "vehicle_telemetry")
      |> filter(fn: (r) => r._field == "speed_kmh")
      |> filter(fn: (r) => r._value == 0.0)
      |> group(columns: ["vehicle_id", "driver", "route_zone"])
      |> count()
      |> map(fn: (r) => ({{r with idle_minutes: r._value / 2}}))
      |> group()
      |> sort(columns: ["idle_minutes"], desc: true)
    '''

    with get_client() as client:
        query_api = client.query_api()
        tables = query_api.query(query, org=INFLUX_ORG)

        print(f"\n{'Vehicle':<12} {'Driver':<10} {'Zone':<8} {'Idle Time (min)':<18} {'Status'}")
        print("-" * 60)
        for table in tables:
            for record in table.records:
                vehicle = record.values.get("vehicle_id", "N/A")
                driver = record.values.get("driver", "N/A")
                zone = record.values.get("route_zone", "N/A")
                idle_min = record.values.get("idle_minutes", 0)
                status = "EXCESSIVE" if idle_min > 40 else "NORMAL"
                print(f"{vehicle:<12} {driver:<10} {zone:<8} {idle_min:<18.0f} {status}")

    print("\n>> RECOMMENDATION: Drivers with EXCESSIVE idle time may need")
    print("   route optimization or dispatch prioritization.")


def main():
    print("SwiftRoute Logistics - Fleet Analytics Dashboard")
    print("Querying Timestream for InfluxDB...")

    query_sla_breach_risk()
    query_fuel_efficiency()
    query_idle_time()

    print("\n" + "=" * 60)
    print("All reports generated successfully.")
    print("=" * 60)


if __name__ == "__main__":
    main()

You now have a complete fleet analytics dashboard that turns raw telemetry into actionable business intelligence. Next up, you will extend this system with a delivery throughput analysis in the Secret Mission.

Secret mission

Delivery Throughput and Staffing Gap Detection

Your three reports tell SwiftRoute what's going wrong. But operations managers need to know when to act. Can you build a throughput analysis that pinpoints understaffed time windows and recommends how many drivers to add?

Clean Up Your Resources

Clean Up Your Resources

Your Timestream for InfluxDB instance is billed per hour while it exists. There is no pause or stop option for this service, so you either keep it running or delete it entirely.

Cost warning

The swiftroute-fleet instance (db.influx.medium) charges approximately $0.13-0.26 per hour while it exists. If you leave it running overnight, you could accumulate several dollars in charges.

Deleting the instance immediately after completing this project is strongly recommended.

Resources you used:

  • Amazon Timestream for InfluxDB instance (swiftroute-fleet).
  • Security group inbound rule (TCP port 8086).
  • Local swiftroute-fleet project folder on your Desktop.
  • Python influxdb-client package.

Keep everything running

No action needed. Choose this if you plan to continue experimenting with Flux queries or extending the fleet simulator in the next few minutes.

  • Your Timestream for InfluxDB instance will continue to charge approximately $0.13-0.26 per hour.
  • Your local project files and Python package remain in place.
  • The security group rule stays open on port 8086.

Pause - I'll come back to this later

Timestream for InfluxDB does not support stopping or pausing an instance. The only way to stop charges is to delete the instance entirely.

If you want to come back later, delete the instance now and recreate it when you are ready. Your local Python scripts will still work once you spin up a new instance and re-run the ingestion.

  • Delete the Timestream instance using the steps in the Delete tab.
  • Keep your local swiftroute-fleet folder. Your simulator, ingestion, and query scripts are reusable.
  • Keep the influxdb-client package installed so you can reconnect later without reinstalling.

Delete - I don't want to use this again

Remove all project resources and stop all charges. Follow each step below.

  • Delete the Timestream for InfluxDB instance:
  • Open the Amazon Timestream for InfluxDB console.
  • In the navigation pane, choose InfluxDB Databases.
  • Select your swiftroute-fleet instance.
  • Choose Delete.
  • Type confirm in the confirmation box, then choose Delete again.

What about the security group rule?

The inbound rule you added (TCP port 8086) remains in your security group after the instance is deleted. Navigate to the EC2 console, choose Security Groups, select the group you modified, click Edit inbound rules, and delete the port 8086 rule.

  • Uninstall the Python package by running:
pip uninstall influxdb-client -y
  • Delete the local project folder by running:
cd ~/Desktop
rm -rf swiftroute-fleet
  • Confirm the folder is gone by running:
ls ~/Desktop

You should no longer see swiftroute-fleet in the output. All charges have stopped and all project files are removed.

Nice Work!

Nice Work!

You just built a production-style fleet telemetry analytics backend for SwiftRoute Logistics using Amazon Timestream for InfluxDB. From simulating 10 delivery vans to running business-critical Flux queries, you now have hands-on experience with time-series data at scale.

You've learned how to:

  • Built a fleet telemetry simulator generating realistic GPS, speed, fuel, and delivery status data for 10 vehicles across 4 hours of route activity.
  • Created an Amazon Timestream for InfluxDB instance and ingested ~4800 time-series data points using the influxdb-client Python library with batch writes.
  • Wrote Flux queries that answer real supply chain business questions including SLA breach detection, fuel efficiency rankings, and idle time analysis.
  • Secret Mission: Built delivery throughput analysis with hourly staffing gap detection across 30-minute time windows.

Ready to quiz yourself?