Python SDK

pip install airpinpoint

from airpinpoint import AirPinpoint
 
client = AirPinpoint(api_key="your-api-key")
 
# List all trackable devices
page = client.trackables.list()
for trackable in page.data:
    print(f"{trackable.name}: enabled={trackable.enabled}")
 
# Get current location
location = client.trackables.current_location(page.data[0].id)
print(f"Lat: {location.latitude}, Lng: {location.longitude}")

from airpinpoint import AsyncAirPinpoint
 
async def main():
    async with AsyncAirPinpoint(api_key="your-api-key") as client:
        page = await client.trackables.list()
        for trackable in page.data:
            print(trackable.name)
 
        # Async auto-pagination
        async for trackable in client.trackables.list():
            print(trackable.name)

# Sync context manager
with AirPinpoint(api_key="your-api-key", timeout=60.0) as client:
    page = client.trackables.list()
 
# Custom base URL for self-hosted deployments
client = AirPinpoint(
    api_key="your-api-key",
    base_url="https://your-instance.example.com/v1",
    default_headers={"X-Custom-Header": "value"},
)

page = client.trackables.list(skip=0, limit=20)
# Returns SyncPage[TrackableDetail]
 
for trackable in page.data:
    print(f"{trackable.name} ({trackable.model}): enabled={trackable.enabled}")

trackable = client.trackables.retrieve("trk_abc123")
print(f"{trackable.name}, paired at {trackable.paired_at}")

location = client.trackables.current_location("trk_abc123")
print(f"Delivery Van #3 is at {location.latitude}, {location.longitude}")
print(f"Accuracy: {location.horizontal_accuracy}m")
print(f"Last seen: {location.timestamp}")

from datetime import datetime, timedelta, timezone
 
end = datetime.now(timezone.utc)
start = end - timedelta(days=7)
 
page = client.trackables.location_history(
    "trk_abc123",
    start_time=start,
    end_time=end,
    limit=100,
)
 
for loc in page.data:
    print(f"{loc.timestamp}: ({loc.latitude}, {loc.longitude})")

battery = client.trackables.battery("trk_abc123")
print(f"Level: {battery.battery_level}")  # 0=full, 1=medium, 2=low, 3=critical
print(f"Days remaining: {battery.estimated_days_remaining}")

result = client.trackables.reset_battery("trk_abc123", battery_months=12)
print(result.message)  # "Battery reset successfully"

page = client.geofences.list(skip=0, limit=20)
for fence in page.data:
    print(f"{fence.name}: radius={fence.radius}m")
 
# Filter by trackable
page = client.geofences.list(trackable_id="trk_abc123")

geofence = client.geofences.create(
    name="Warehouse Zone A",
    latitude=37.7749,
    longitude=-122.4194,
    radius=200,
    trackable_id=["trk_abc123", "trk_def456"],
)
print(f"Created geofence: {geofence.id}")

updated = client.geofences.update(
    "geofence_789012",
    name="Warehouse Zone A (Expanded)",
    radius=350,
)
print(f"Updated radius to {updated.radius}m")

client.geofences.delete("geofence_789012")

result = client.geofences.test_webhook("geofence_789012", event_type="entry")
print(f"Delivery ID: {result.delivery_id}")

# List recent deliveries
deliveries = client.geofences.webhooks.list(
    "geofence_789012",
    limit=10,
    successful=True,
)
for d in deliveries.data:
    print(f"{d.id}: status={d.status_code}, at={d.created_at}")
 
# Get a specific delivery
delivery = client.geofences.webhooks.retrieve("geofence_789012", "whd_abc123")
print(f"Response body: {delivery.response_body}")

from airpinpoint import AirPinpoint
 
client = AirPinpoint(api_key="your-api-key")
 
# 1. List devices to get IDs
page = client.trackables.list()
forklift_ids = [t.id for t in page.data if "Forklift" in t.name]
 
# 2. Create a geofence around the loading dock
geofence = client.geofences.create(
    name="Loading Dock B",
    latitude=37.7749,
    longitude=-122.4194,
    radius=150,
    trackable_id=forklift_ids,
)
 
# 3. Test the webhook to make sure it works
result = client.geofences.test_webhook(geofence.id, event_type="entry")
print(f"Test delivery: {result.delivery_id}")
 
# 4. Later, expand the zone
client.geofences.update(geofence.id, radius=250)
 
# 5. Clean up when no longer needed
client.geofences.delete(geofence.id)

link = client.share_links.create(trackable_id="trk_abc123", hours=24)
print(f"Share URL: {link.url}")  # valid for 24 hours

account = client.account.retrieve()
print(f"Org: {account.organization_name}")
print(f"Plan: {account.plan}")
print(f"Devices: {account.device_count}/{account.device_limit}")

from datetime import date
 
usage = client.usage.retrieve(
    start_date=date(2025, 1, 1),
    end_date=date(2025, 1, 31),
)
print(f"API calls: {usage.api_calls}")
print(f"Location fetches: {usage.location_fetches}")
 
# Get totals for a billing period
totals = client.usage.total(
    start_date=date(2025, 1, 1),
    end_date=date(2025, 1, 31),
)
print(f"Total API calls: {totals.total_api_calls}")

page = client.trackables.list(limit=10)
print(page.data)       # list of TrackableDetail objects
print(page.has_more)   # True if more pages exist
print(page.total)      # total count across all pages

# Sync: iterates through all pages automatically
for trackable in client.trackables.list():
    print(trackable.name)
 
# Async: same behavior with async for
async for trackable in client.trackables.list():
    print(trackable.name)

from airpinpoint import AirPinpoint, APIError, NotFoundError, RateLimitError
 
client = AirPinpoint(api_key="your-api-key")
 
try:
    trackable = client.trackables.retrieve("trk_nonexistent")
except NotFoundError:
    print("Device not found. Check the trackable ID.")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.response.headers.get('Retry-After')}s")
except APIError as e:
    print(f"API error {e.status_code}: {e.message}")

from airpinpoint import Webhook, WebhookSignatureError
 
try:
    event = Webhook.construct_event(
        payload=request.body,
        signature=request.headers["X-AirPinpoint-Signature"],
        secret="whsec_your_secret",
    )
    if event.event == "geofence.entry":
        print(f"{event.beacon.name} entered {event.geofence.name}")
    elif event.event == "geofence.exit":
        print(f"{event.beacon.name} left {event.geofence.name}")
except WebhookSignatureError:
    print("Invalid signature")

from flask import Flask, request, Response
from airpinpoint import Webhook, WebhookSignatureError
 
app = Flask(__name__)
WEBHOOK_SECRET = "whsec_your_secret"
 
@app.route("/webhooks/geofence", methods=["POST"])
def handle_geofence_webhook():
    try:
        event = Webhook.construct_event(
            payload=request.get_data(as_text=True),
            signature=request.headers["X-AirPinpoint-Signature"],
            secret=WEBHOOK_SECRET,
        )
    except WebhookSignatureError:
        return Response("Invalid signature", status=400)
 
    if event.event == "geofence.entry":
        notify_dispatch(event.beacon.name, event.geofence.name)
 
    return Response("OK", status=200)

from fastapi import FastAPI, Request, HTTPException
from airpinpoint import Webhook, WebhookSignatureError
 
app = FastAPI()
WEBHOOK_SECRET = "whsec_your_secret"
 
@app.post("/webhooks/geofence")
async def handle_geofence_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-AirPinpoint-Signature", "")
 
    try:
        event = Webhook.construct_event(
            payload=body.decode(),
            signature=signature,
            secret=WEBHOOK_SECRET,
        )
    except WebhookSignatureError:
        raise HTTPException(status_code=400, detail="Invalid signature")
 
    if event.event == "geofence.exit":
        print(f"{event.beacon.name} left {event.geofence.name}")
 
    return {"status": "ok"}

from airpinpoint.utils import distance, bearing, is_inside_geofence
 
# How far is Forklift #42 from the warehouse?
dist = distance(forklift_location, warehouse_location, unit="meters")
print(f"Distance: {dist:.1f}m")
 
# Which direction is it heading?
heading = bearing(previous_location, forklift_location)
print(f"Bearing: {heading:.0f} degrees")
 
# Is it inside the geofence?
inside = is_inside_geofence(forklift_location, warehouse_geofence)
print(f"Inside warehouse zone: {inside}")

from airpinpoint.utils import detect_stops, dwell_time
 
# Find where Delivery Van #3 stopped for more than 5 minutes
history = client.trackables.location_history(
    "trk_abc123",
    start_time=start,
    end_time=end,
)
stops = detect_stops(list(history), min_duration_s=300, radius_m=50)
for stop in stops:
    print(f"Stopped at ({stop.latitude}, {stop.longitude}) "
          f"for {stop.duration_s / 60:.0f} min")
 
# How long did it spend at the job site today?
minutes = dwell_time(
    list(history),
    center_lat=37.7749,
    center_lng=-122.4194,
    radius_m=200,
)
print(f"Time on site: {minutes / 60:.1f} hours")

from airpinpoint.utils import to_geojson, encode_polyline
import json
 
history = list(client.trackables.location_history(
    "trk_abc123",
    start_time=start,
    end_time=end,
))
 
# Export as GeoJSON for mapping tools
geojson = to_geojson(history)
with open("route.geojson", "w") as f:
    json.dump(geojson, f)
 
# Encode as polyline for Google Maps URLs
polyline = encode_polyline(history)
print(f"Polyline: {polyline}")

from airpinpoint.utils import (
    filter_by_accuracy,
    filter_by_speed,
    deduplicate,
)
 
# Clean up raw location history before analysis
raw = list(client.trackables.location_history(
    "trk_abc123",
    start_time=start,
    end_time=end,
    limit=500,
))
 
# Pipeline: accuracy filter -> speed filter -> dedup
clean = filter_by_accuracy(raw, max_accuracy_m=30)
clean = filter_by_speed(clean, max_speed_kmh=150)
clean = deduplicate(clean, radius_m=10, time_window_s=120)
 
print(f"Kept {len(clean)}/{len(raw)} locations after cleanup")

from airpinpoint.utils import estimate_battery_days_remaining, battery_health_status
 
# Monitor battery health across your fleet
page = client.trackables.list()
for trackable in page.data:
    battery = client.trackables.battery(trackable.id)
    days = estimate_battery_days_remaining(battery)
    status = battery_health_status(battery)
 
    if status != "good":
        print(f"[{status.upper()}] {trackable.name}: ~{days} days remaining")

# Type hints work with all models
from airpinpoint.types import TrackableDetail, LocationBase
 
def process_device(device: TrackableDetail) -> None:
    loc: LocationBase = client.trackables.current_location(device.id)
    print(f"{device.name}: {loc.latitude}, {loc.longitude}")