Asset Management

Intelligent quota monitoring, asset activation, and download management

The Asset Management module provides comprehensive capabilities for managing Planet subscription quotas, activating assets, and downloading PlanetScope imagery with intelligent progress tracking and user confirmation workflows.

Overview

The AssetManager handles the complete asset lifecycle including:

• Quota Monitoring: Real-time tracking of subscription usage across multiple Planet APIs
• Asset Activation: Automated asset processing with status tracking
• Download Management: Parallel downloads with retry logic and progress tracking
• User Confirmation: Interactive workflows with impact assessment
• ROI Clipping: Automatic scene clipping to regions of interest

Quick Start

Basic Asset Management

from planetscope_py import AssetManager, PlanetScopeQuery

# Initialize with authentication
query = PlanetScopeQuery()
asset_manager = AssetManager(query.auth)

# Check current quota usage
quota_info = await asset_manager.get_quota_info()
print(f"Current usage: {quota_info.used_area_km2:.1f} / {quota_info.limit_area_km2} km²")
print(f"Remaining: {quota_info.remaining_area_km2:.1f} km²")
print(f"Usage: {quota_info.usage_percentage:.1%}")

# Download assets with confirmation
selected_scenes = results['features'][:10]  # First 10 scenes
downloads = await asset_manager.activate_and_download_assets(
    scenes=selected_scenes,
    asset_types=["ortho_analytic_4b"],
    output_dir="downloads",
    clip_to_roi=roi
)

print(f"Downloaded {len(downloads)} assets")

Quota Management

Checking Quota Usage

# Get comprehensive quota information
quota_info = await asset_manager.get_quota_info()

print("Quota Information:")
print(f"  Monthly Limit: {quota_info.monthly_limit_km2:,.0f} km²")
print(f"  Current Usage: {quota_info.current_usage_km2:,.0f} km²")
print(f"  Remaining: {quota_info.remaining_area_km2:,.0f} km²")
print(f"  Usage Percentage: {quota_info.usage_percentage:.1%}")

# Check if near limit
if quota_info.is_near_limit:
    print("⚠️  WARNING: Usage is near monthly limit")

# Check if download is possible
download_estimate = asset_manager.calculate_download_impact(selected_scenes, roi)
if quota_info.can_download(download_estimate['area_km2']):
    print("✅ Download is within quota limits")
else:
    print("❌ Download would exceed quota limits")

Download Impact Calculation

# Calculate impact before downloading
scenes_to_download = results['features'][:20]
roi = box(9.04, 45.40, 9.28, 45.52)

impact = asset_manager.calculate_download_impact(scenes_to_download, roi)

print("Download Impact Analysis:")
print(f"  Number of scenes: {impact['scene_count']}")
print(f"  Total area: {impact['area_km2']:.2f} km²")
print(f"  Average scene size: {impact['avg_scene_area_km2']:.2f} km²")
print(f"  Estimated file size: {impact['estimated_file_size_gb']:.1f} GB")

# Check quota impact
print(f"  Current usage: {quota_info.current_usage_km2:.1f} km²")
print(f"  After download: {quota_info.current_usage_km2 + impact['area_km2']:.1f} km²")
print(f"  Remaining after: {quota_info.remaining_area_km2 - impact['area_km2']:.1f} km²")

Asset Activation and Download

Basic Download Workflow

# Complete download workflow with user confirmation
downloads = await asset_manager.activate_and_download_assets(
    scenes=selected_scenes,
    asset_types=["ortho_analytic_4b", "ortho_visual"],  # Multiple asset types
    output_dir="planet_downloads",
    clip_to_roi=roi,
    confirm_download=True,  # Ask for user confirmation
    max_concurrent=3  # Limit concurrent downloads
)

# Check download results
successful_downloads = [d for d in downloads if d.status == AssetStatus.COMPLETED]
failed_downloads = [d for d in downloads if d.status == AssetStatus.FAILED]

print(f"Successful: {len(successful_downloads)}")
print(f"Failed: {len(failed_downloads)}")

Advanced Download Configuration

# Configure download behavior
config = {
    'max_concurrent_downloads': 5,
    'download_chunk_size': 16384,  # 16KB chunks
    'retry_attempts': 3,
    'retry_delay': 5.0,  # seconds
    'timeout': 300,  # 5 minutes per download
    'rate_limits': {
        'activation': 5,   # 5 activations per second
        'download': 10     # 10 downloads per second
    }
}

asset_manager = AssetManager(query.auth, config)

# Download with custom configuration
downloads = await asset_manager.activate_and_download_assets(
    scenes=scenes,
    asset_types=["ortho_analytic_4b"],
    output_dir="downloads",
    max_concurrent=config['max_concurrent_downloads']
)

Progress Tracking

# Set up progress callback
def download_progress_callback(job: DownloadJob):
    """Custom progress tracking callback."""
    if job.progress_percentage > 0:
        print(f"{job.scene_id}: {job.progress_percentage:.1f}% - {job.status.value}")

asset_manager.progress_callback = download_progress_callback

# Download with progress tracking
downloads = await asset_manager.activate_and_download_assets(
    scenes=scenes,
    asset_types=["ortho_analytic_4b"]
)

User Confirmation Workflows

Interactive Confirmation

# The system automatically shows confirmation dialog:
"""
==================================================
DOWNLOAD CONFIRMATION
==================================================
Scenes to download: 15
Asset types: ['ortho_analytic_4b']
Total estimated area: 245.67 km²
Estimated file size: 3.2 GB

Current Quota Status:
  Used: 1,234 / 3,000 km² (41.1%)
  Remaining: 1,766 km²
  Required: 245.67 km²
  Available after download: 1,520.33 km²

==================================================
Proceed with download? (y/n):
"""

# Automatic approval for scripted workflows
downloads = await asset_manager.activate_and_download_assets(
    scenes=scenes,
    confirm_download=False  # Skip confirmation
)

Custom Confirmation Logic

# Implement custom confirmation logic
def custom_confirmation_logic(quota_info, download_impact):
    """Custom logic for download approval."""
    # Always approve small downloads
    if download_impact['area_km2'] < 50:
        return True
    
    # Check business hours
    from datetime import datetime
    if datetime.now().hour < 9 or datetime.now().hour > 17:
        print("Large downloads only during business hours")
        return False
    
    # Check remaining quota percentage
    remaining_after = quota_info.remaining_area_km2 - download_impact['area_km2']
    remaining_percentage = remaining_after / quota_info.monthly_limit_km2
    
    if remaining_percentage < 0.2:  # Keep 20% buffer
        print("Download would leave less than 20% quota remaining")
        return False
    
    return True

# Override confirmation method
asset_manager.get_user_confirmation = custom_confirmation_logic

Asset Types and Formats

Available Asset Types

# Common PlanetScope asset types
asset_types = [
    "ortho_analytic_4b",    # 4-band analytic (Blue, Green, Red, NIR)
    "ortho_analytic_8b",    # 8-band analytic (Coastal, Blue, Green I, Green, Yellow, Red, Red Edge, NIR)
    "ortho_visual",         # 3-band visual (RGB)
    "ortho_analytic_sr",    # Surface reflectance
    "ortho_udm2"           # Usable data mask
]

# Download multiple asset types
downloads = await asset_manager.activate_and_download_assets(
    scenes=scenes,
    asset_types=["ortho_analytic_4b", "ortho_visual", "ortho_udm2"]
)

# Organize downloads by asset type
by_asset_type = {}
for download in downloads:
    asset_type = download.asset_type
    if asset_type not in by_asset_type:
        by_asset_type[asset_type] = []
    by_asset_type[asset_type].append(download)

for asset_type, jobs in by_asset_type.items():
    successful = [j for j in jobs if j.status == AssetStatus.COMPLETED]
    print(f"{asset_type}: {len(successful)}/{len(jobs)} successful")

ROI Clipping

Automatic Clipping During Download

from shapely.geometry import box

# Define region of interest
roi = box(9.04, 45.40, 9.28, 45.52)  # Milan center

# Download with automatic ROI clipping
downloads = await asset_manager.activate_and_download_assets(
    scenes=scenes,
    asset_types=["ortho_analytic_4b"],
    clip_to_roi=roi,  # Enable ROI clipping
    output_dir="clipped_downloads"
)

# Clipped files are automatically saved with "_clipped" suffix
for download in downloads:
    if download.status == AssetStatus.COMPLETED:
        print(f"Original: {download.original_file_path}")
        print(f"Clipped: {download.clipped_file_path}")

Custom Clipping Configuration

# Configure clipping behavior
clipping_config = {
    'clip_to_roi': True,
    'buffer_meters': 100,  # Add 100m buffer around ROI
    'output_format': 'GTiff',
    'compression': 'LZW',
    'create_overview': True
}

# Apply custom clipping configuration
downloads = await asset_manager.activate_and_download_assets(
    scenes=scenes,
    clip_to_roi=roi,
    clipping_config=clipping_config
)

Download Job Management

DownloadJob Status Tracking

from planetscope_py import AssetStatus

# Monitor download job status
for job in downloads:
    print(f"Scene: {job.scene_id}")
    print(f"Asset: {job.asset_type}")
    print(f"Status: {job.status.value}")
    print(f"Progress: {job.progress_percentage:.1f}%")
    
    if job.status == AssetStatus.COMPLETED:
        print(f"File: {job.downloaded_file_path}")
        print(f"Size: {job.file_size_mb:.1f} MB")
    elif job.status == AssetStatus.FAILED:
        print(f"Error: {job.error_message}")
    
    print("---")

Retry Failed Downloads

# Identify failed downloads
failed_jobs = [job for job in downloads if job.status == AssetStatus.FAILED]

if failed_jobs:
    print(f"Retrying {len(failed_jobs)} failed downloads...")
    
    # Retry failed downloads
    retry_downloads = await asset_manager.retry_failed_downloads(failed_jobs)
    
    # Check retry results
    retry_successful = [d for d in retry_downloads if d.status == AssetStatus.COMPLETED]
    print(f"Retry successful: {len(retry_successful)}/{len(failed_jobs)}")

Error Handling

Common Download Errors

from planetscope_py.exceptions import AssetError

try:
    downloads = await asset_manager.activate_and_download_assets(scenes=scenes)
    
except AssetError as e:
    print(f"Asset management error: {e}")
    
    # Handle specific error types
    if "quota exceeded" in str(e).lower():
        print("Quota limit reached - try fewer scenes")
    elif "activation failed" in str(e).lower():
        print("Asset activation failed - check scene availability")
    elif "download failed" in str(e).lower():
        print("Download failed - check network connection")

except Exception as e:
    print(f"Unexpected error: {e}")

Robust Download with Error Recovery

async def robust_download_workflow(asset_manager, scenes, roi):
    """Robust download workflow with comprehensive error handling."""
    try:
        # Check quota first
        quota_info = await asset_manager.get_quota_info()
        download_impact = asset_manager.calculate_download_impact(scenes, roi)
        
        if not quota_info.can_download(download_impact['area_km2']):
            print("Insufficient quota - reducing scene count")
            max_scenes = int(quota_info.remaining_area_km2 / download_impact['avg_scene_area_km2'])
            scenes = scenes[:max_scenes]
        
        # Start download
        downloads = await asset_manager.activate_and_download_assets(
            scenes=scenes,
            clip_to_roi=roi,
            confirm_download=False
        )
        
        # Handle partial failures
        successful = [d for d in downloads if d.status == AssetStatus.COMPLETED]
        failed = [d for d in downloads if d.status == AssetStatus.FAILED]
        
        if failed:
            print(f"Initial download: {len(successful)}/{len(downloads)} successful")
            print("Retrying failed downloads...")
            
            retry_downloads = await asset_manager.retry_failed_downloads(failed)
            retry_successful = [d for d in retry_downloads if d.status == AssetStatus.COMPLETED]
            
            total_successful = len(successful) + len(retry_successful)
            print(f"Final result: {total_successful}/{len(downloads)} successful")
        
        return successful + retry_successful
        
    except AssetError as e:
        print(f"Asset management error: {e}")
        return []

Performance Optimization

Concurrent Download Management

# Optimize for different scenarios

# For fast connections - increase concurrency
fast_config = {
    'max_concurrent_downloads': 10,
    'download_chunk_size': 32768  # 32KB chunks
}

# For slow connections - reduce concurrency
slow_config = {
    'max_concurrent_downloads': 2,
    'download_chunk_size': 8192   # 8KB chunks
}

# For large files - increase chunk size
large_file_config = {
    'max_concurrent_downloads': 5,
    'download_chunk_size': 65536  # 64KB chunks
}

Memory Management

# For memory-constrained environments
memory_efficient_config = {
    'max_concurrent_downloads': 2,  # Reduce concurrency
    'download_chunk_size': 4096,    # Smaller chunks
    'cleanup_temp_files': True      # Auto-cleanup
}

asset_manager = AssetManager(auth, memory_efficient_config)

# Process in batches for large scene lists
def batch_download(scenes, batch_size=50):
    """Download scenes in batches to manage memory."""
    results = []
    
    for i in range(0, len(scenes), batch_size):
        batch = scenes[i:i + batch_size]
        print(f"Processing batch {i//batch_size + 1}: {len(batch)} scenes")
        
        batch_downloads = await asset_manager.activate_and_download_assets(
            scenes=batch,
            confirm_download=False
        )
        results.extend(batch_downloads)
    
    return results

Integration with Other Components

With Scene Discovery

from planetscope_py import PlanetScopeQuery

# Integrated workflow: search and download
query = PlanetScopeQuery()
asset_manager = AssetManager(query.auth)

# Search for scenes
results = query.search_scenes(
    geometry=roi,
    start_date="2024-01-01",
    end_date="2024-06-30",
    cloud_cover_max=0.2
)

# Quality filter before download
high_quality_scenes = []
for scene in results['features']:
    props = scene['properties']
    if props.get('sun_elevation', 0) > 30:  # Good sun angle
        high_quality_scenes.append(scene)

# Download high-quality scenes
downloads = await asset_manager.activate_and_download_assets(
    scenes=high_quality_scenes,
    clip_to_roi=roi
)

With GeoPackage Export

from planetscope_py import GeoPackageManager

# Download and immediately export to GeoPackage
downloads = await asset_manager.activate_and_download_assets(scenes=scenes)

# Get downloaded file paths
downloaded_files = [
    d.downloaded_file_path for d in downloads 
    if d.status == AssetStatus.COMPLETED
]

# Create GeoPackage with imagery
geopackage_manager = GeoPackageManager()
geopackage_manager.create_scene_geopackage(
    scenes=scenes,
    output_path="analysis_with_imagery.gpkg",
    roi=roi,
    downloaded_files=downloaded_files
)

Best Practices

1. Monitor Quota Usage

# Regular quota monitoring
async def monitor_quota_usage():
    quota_info = await asset_manager.get_quota_info()
    
    print(f"Quota Status: {quota_info.usage_percentage:.1%}")
    
    if quota_info.is_near_limit:
        # Send alert or take action
        print("⚠️  Near quota limit - consider upgrading subscription")
    
    return quota_info

# Check quota before major operations
quota_info = await monitor_quota_usage()

2. Efficient Scene Selection

# Pre-filter scenes to optimize quota usage
def optimize_scene_selection(scenes, roi, max_area_km2):
    """Select scenes efficiently to stay within quota."""
    from planetscope_py.utils import calculate_area_km2
    
    # Calculate area for each scene
    scene_areas = []
    for scene in scenes:
        geom = shape(scene['geometry'])
        area = calculate_area_km2(geom)
        scene_areas.append((scene, area))
    
    # Sort by area (prefer smaller scenes for better coverage)
    scene_areas.sort(key=lambda x: x[1])
    
    # Select scenes up to quota limit
    selected_scenes = []
    total_area = 0
    
    for scene, area in scene_areas:
        if total_area + area <= max_area_km2:
            selected_scenes.append(scene)
            total_area += area
        else:
            break
    
    return selected_scenes

# Apply optimization
optimized_scenes = optimize_scene_selection(
    scenes=results['features'],
    roi=roi,
    max_area_km2=quota_info.remaining_area_km2 * 0.8  # Use 80% of remaining
)

3. Automated Download Management

# Set up automated download management
class AutomatedDownloadManager:
    def __init__(self, asset_manager):
        self.asset_manager = asset_manager
        self.download_queue = []
        self.max_daily_quota = 500  # km²
        
    async def add_to_queue(self, scenes, priority="normal"):
        """Add scenes to download queue with priority."""
        self.download_queue.append({
            'scenes': scenes,
            'priority': priority,
            'added_at': datetime.now()
        })
    
    async def process_queue(self):
        """Process download queue based on quota availability."""
        quota_info = await self.asset_manager.get_quota_info()
        
        if quota_info.remaining_area_km2 < 100:  # Less than 100 km² remaining
            print("Quota too low - skipping downloads")
            return
        
        # Sort queue by priority
        self.download_queue.sort(
            key=lambda x: (x['priority'] == 'high', -x['added_at'].timestamp())
        )
        
        for item in self.download_queue[:]:
            impact = self.asset_manager.calculate_download_impact(item['scenes'])
            
            if impact['area_km2'] <= quota_info.remaining_area_km2:
                print(f"Processing {len(item['scenes'])} scenes...")
                downloads = await self.asset_manager.activate_and_download_assets(
                    scenes=item['scenes'],
                    confirm_download=False
                )
                self.download_queue.remove(item)
                
                # Update quota info
                quota_info = await self.asset_manager.get_quota_info()
            else:
                print("Insufficient quota for remaining items")
                break

# Usage
auto_manager = AutomatedDownloadManager(asset_manager)
await auto_manager.add_to_queue(high_priority_scenes, priority="high")
await auto_manager.add_to_queue(regular_scenes, priority="normal")
await auto_manager.process_queue()

Troubleshooting

Common Issues

1. Quota exceeded errors

# Check quota before operations
quota_info = await asset_manager.get_quota_info()
if quota_info.usage_percentage > 0.9:
    print("Quota usage > 90% - consider reducing download scope")

2. Download timeouts

# Increase timeout for slow connections
config = {
    'timeout': 600,  # 10 minutes
    'retry_attempts': 5,
    'retry_delay': 10.0
}
asset_manager = AssetManager(auth, config)

3. Asset activation failures

# Check asset availability before download
async def check_asset_availability(scene_id, asset_type):
    """Check if asset is available for download."""
    try:
        asset_info = await asset_manager.get_asset_info(scene_id, asset_type)
        return asset_info['status'] in ['active', 'activating']
    except:
        return False

# Filter available assets
available_scenes = []
for scene in scenes:
    if await check_asset_availability(scene['properties']['id'], 'ortho_analytic_4b'):
        available_scenes.append(scene)

4. Network connection issues

# Implement connection retry logic
async def robust_asset_manager_operation(operation_func, *args, **kwargs):
    """Robust wrapper for asset manager operations."""
    max_retries = 3
    
    for attempt in range(max_retries):
        try:
            return await operation_func(*args, **kwargs)
        except ConnectionError as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Connection error, retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)
            else:
                raise e

# Usage
downloads = await robust_asset_manager_operation(
    asset_manager.activate_and_download_assets,
    scenes=scenes
)

Examples

See the Asset Management Examples page for detailed workflow examples and use cases.