Use Cases

import requests

API_KEY = "your_api_key"
BASE_URL = "https://datadocked.com/api/vessels_operations"

# Your fleet IMO numbers
fleet_imos = ["9247431", "9184419", "9465411"]

def track_fleet(imos: list) -> list:
    """Track multiple vessels using bulk search"""
    response = requests.get(
        f"{BASE_URL}/get-vessels-location-bulk-search",
        params={"imo_or_mmsi": ",".join(imos)},
        headers={"x-api-key": API_KEY}
    )
    return response.json()["results"]

# Get current positions
vessels = track_fleet(fleet_imos)
for vessel in vessels:
    print(f"{vessel['name']}: {vessel['destination']} (ETA: {vessel['etaUtc']})")

def assess_vessel_risk(imo: str) -> dict:
    """Get inspection and classification data for risk assessment"""
    response = requests.get(
        f"{BASE_URL}/vessel-mou",
        params={"imo": imo},
        headers={"x-api-key": API_KEY}
    )
    data = response.json()["detail"]

    # Extract risk indicators
    inspections = data.get("inspection", [])
    detentions = sum(1 for i in inspections if i.get("detention") == "Yes")
    total_deficiencies = sum(int(i.get("number_of_deficiencies", 0))
                            for i in inspections)

    return {
        "total_inspections": len(inspections),
        "detentions": detentions,
        "total_deficiencies": total_deficiencies,
        "classification": data.get("classification", {}).get("status", {})
    }

risk = assess_vessel_risk("9184419")
print(f"Detentions: {risk['detentions']}, Deficiencies: {risk['total_deficiencies']}")

def get_port_traffic(port_code: str) -> dict:
    """Get current port activity"""
    traffic = {}

    for search_type in ["arrival", "departure", "in_port", "expected"]:
        response = requests.get(
            f"{BASE_URL}/port-calls-by-port",
            params={
                "port_call": port_code,
                "search_type": search_type,
                "page": 1
            },
            headers={"x-api-key": API_KEY}
        )
        data = response.json()["detail"]
        traffic[search_type] = data.get(search_type, {}).get("total", 0)

    return traffic

# Rotterdam port traffic
traffic = get_port_traffic("NLRTM")
print(f"In port: {traffic['in_port']}, Expected: {traffic['expected']}")

from datetime import datetime, timedelta

def get_voyage_history(imo: str, days: int = 30) -> list:
    """Get historical positions for route analysis"""
    end_date = datetime.utcnow()
    start_date = end_date - timedelta(days=days)

    response = requests.get(
        f"{BASE_URL}/get-vessel-historical-data",
        params={
            "imo_or_mmsi": imo,
            "from_date": start_date.isoformat() + "Z",
            "to_date": end_date.isoformat() + "Z"
        },
        headers={"x-api-key": API_KEY}
    )
    return response.json().get("data", [])

# Get 14-day voyage history
positions = get_voyage_history("9321483", days=14)
print(f"Retrieved {len(positions)} positions")

def get_vessel_conditions(imo: str) -> dict:
    """Get vessel location and current weather"""
    # Get location
    location = requests.get(
        f"{BASE_URL}/get-vessel-location",
        params={"imo_or_mmsi": imo},
        headers={"x-api-key": API_KEY}
    ).json()["detail"]

    # Get weather
    weather = requests.get(
        f"{BASE_URL}/get-vessel-weather",
        params={"imo": imo},
        headers={"x-api-key": API_KEY}
    ).json()["detail"]

    return {
        "vessel": location["name"],
        "position": f"{location['latitude']}, {location['longitude']}",
        "speed": location["speed"],
        "temperature": weather.get("temperature"),
        "wind_speed": weather.get("windSpeed"),
        "wave_height": weather.get("waves")
    }

conditions = get_vessel_conditions("8909070")
print(f"Waves: {conditions['wave_height']}, Wind: {conditions['wind_speed']}")

import time

def track_vessel_continuously(imo: str, interval_seconds: int = 300):
    """Poll vessel location at regular intervals"""
    while True:
        location = get_vessel_location(imo)

        # Process update (e.g., store in database, check geofence)
        process_location_update(location)

        time.sleep(interval_seconds)

def process_fleet_batch(imos: list, batch_size: int = 50):
    """Process large fleets in batches"""
    results = []

    for i in range(0, len(imos), batch_size):
        batch = imos[i:i + batch_size]
        batch_results = track_fleet(batch)
        results.extend(batch_results)
        time.sleep(1)  # Rate limiting

    return results