Examples¶

This page contains examples of how to use the Pixel client library for various tasks. The examples demonstrate both synchronous and asynchronous usage of the client.

Synchronous Examples¶

Working with Projects¶

This example demonstrates how to use the synchronous client to work with projects:

from pixel_client import get_client
from geojson_pydantic import Polygon

client = get_client()

# Example: List projects
projects = client.list_projects()
for project in projects:
    print(f"Project ID: {project['id']}, Name: {project['name']}")

# Example: Create a new project
new_project = client.create_project(
    name="My New Project",
    description="A test project",
    area_of_interest=Polygon.model_validate(
        {
            "type": "Polygon",
            "coordinates": [[[-10, 10], [-10, 20], [0, 20], [0, 10], [-10, 10]]],
        }
    ),
    tags=["test", "example"],
)
print(f"New project created: {new_project}")

Working with Data Collections¶

This example shows how to create data collections, upload images, and optimize rasters using the synchronous client:

from pathlib import Path
from pixel_client import get_client
from pixel_client.models import PixelUploadFile, NearblackOptions

# Initialize the synchronous client
client = get_client()

# Example: Create a data collection
project_id = 123  # Replace with your actual project ID
data_collection = client.create_data_collection(
    project_id=project_id,
    name="My Data Collection",
    description="An example data collection",
    data_collection_type="image",
    tags=["example", "images"],
)
print(f"New data collection created: {data_collection}")

# Example: List data collections in a project
data_collections = client.list_data_collections(project_id)
for dc in data_collections:
    print(
        f"Data Collection ID: {dc['id']}, Name: {dc['name']}, Type: {dc['data_collection_type']}"
    )

# Example: Upload images to a data collection
image_files = [
    PixelUploadFile(file=Path("path/to/image1.jpg")),
    PixelUploadFile(file=Path("path/to/image2.jpg")),
]

uploaded_images, errors = client.upload_multiple_images(
    project_id=project_id, data_collection_id=data_collection["id"], files=image_files
)
for image in uploaded_images:
    print(f"Uploaded image: {image['name']}")
for error in errors:
    print(f"Error uploading image: {error}")

# Example: Optimize rasters in a data collection
optimized_rasters = client.optimize_rasters(
    project_id=project_id,
    data_collection_id=data_collection["id"],
    profile="rgb",
    nearblack=NearblackOptions(enabled=True, color="black"),
    overview_resampling="average",
)

for raster in optimized_rasters:
    print(
        f"Optimized raster: {raster['raster_id']}, Profile: {raster['profile']}, Status: {raster['status']}"
    )

Asynchronous Examples¶

Working with Projects¶

This example demonstrates how to use the asynchronous client to work with projects:

import asyncio
from pixel_client import get_client
from geojson_pydantic import Polygon


async def main():
    async_client = get_client(async_=True)

    # Example: List projects
    projects = await async_client.list_projects()
    for project in projects:
        print(f"Project ID: {project['id']}, Name: {project['name']}")

    # Example: Create a new project
    new_project = await async_client.create_project(
        name="My New Async Project",
        description="A test async project",
        area_of_interest=Polygon.model_validate(
            {
                "type": "Polygon",
                "coordinates": [[[-10, 10], [-10, 20], [0, 20], [0, 10], [-10, 10]]],
            }
        ),
        tags=["test", "async"],
    )
    print(f"New async project created: {new_project}")


asyncio.run(main())

Working with Data Collections¶

This example shows how to create data collections, upload images, and optimize rasters using the asynchronous client:

import asyncio
from pathlib import Path
from pixel_client import get_client
from pixel_client.models import PixelUploadFile, NearblackOptions


async def main():
    # Initialize the asynchronous client
    client = get_client(async_=True)

    # Example: Create a data collection
    project_id = 123  # Replace with your actual project ID
    data_collection = await client.create_data_collection(
        project_id=project_id,
        name="My Async Data Collection",
        description="An example async data collection",
        data_collection_type="image",
        tags=["example", "async", "images"],
    )
    print(f"New async data collection created: {data_collection}")

    # Example: List data collections in a project
    data_collections = await client.list_data_collections(project_id)
    for dc in data_collections:
        print(
            f"Async Data Collection ID: {dc['id']}, Name: {dc['name']}, Type: {dc['data_collection_type']}"
        )

    # Example: Upload images to a data collection
    image_files = [
        PixelUploadFile(file=Path("path/to/image1.jpg")),
        PixelUploadFile(file=Path("path/to/image2.jpg")),
    ]

    uploaded_images, errors = await client.upload_multiple_images(
        project_id=project_id,
        data_collection_id=data_collection["id"],
        files=image_files,
    )
    for image in uploaded_images:
        print(f"Uploaded async image: {image['name']}")
    for error in errors:
        print(f"Error uploading async image: {error}")

    # Example: Optimize rasters in a data collection
    optimized_rasters = await client.optimize_rasters(
        project_id=project_id,
        data_collection_id=data_collection["id"],
        profile="rgb",
        nearblack=NearblackOptions(enabled=True, color="black"),
        overview_resampling="average",
    )

    for raster in optimized_rasters:
        print(
            f"Optimized async raster: {raster['raster_id']}, Profile: {raster['profile']}, Status: {raster['status']}"
        )


if __name__ == "__main__":
    asyncio.run(main())

These examples demonstrate some of the basic operations you can perform using the Pixel client library, both synchronously and asynchronously. For more detailed information on available methods and advanced usage, please refer to the API Reference section.

Complex Workflows¶

Raster Upload, Optimization, and ArcGIS Service Deployment¶

This example demonstrates a complete workflow for:

Finding or creating a raster data collection (RGB, DTM, or DSM) if it doesn't exist
Uploading raster files from a folder structure
Optimizing the rasters with the appropriate profile (RGB or terrain)
Creating or updating an ArcGIS Image Service

The script is designed as a command-line tool that can be used to automate the process of updating raster data in ArcGIS services. It supports different types of raster data collections (RGB, DTM, DSM) and applies the appropriate optimization profile based on the data collection type.

Raster Workflow Script

Copy this script if you want to give it a try

import datetime
from pathlib import Path
from typing import Iterator, Any, Literal

from pixel_client import get_client, PixelClient


from pixel_client.models import (
    RasterInfo,
    PixelUploadFile,
    MetadataObject,
    RasterMetadataFields,
    ArcgisServiceCreateInput,
    ArcgisServiceUpdateInput,
    ArcgisServiceCreateOptions,
    ArcGISImageServiceOptions,
    ArcGISImageServiceWMSOptions,
)


# Constants
SUPPORTED_COLLECTION_TYPES = ["RGB", "DTM", "DSM"]
DEFAULT_SRID = 25832
MULTIPART_PART_SIZE = 50 * 1024 * 1024  # 50 MB part size
TERRAIN_PROFILE = "terrain"
RGB_PROFILE = "rgb"


def iter_raster_files(raster_folder: Path) -> Iterator[PixelUploadFile]:
    """
    Iterate through a folder of raster files and yield PixelUploadFile objects.

    This function looks for .tif files with the naming pattern <some_name>_YYYY-MM-DD.tif
    and their associated support files (.tfw, .aux.xml, etc.) and creates PixelUploadFile
    objects with appropriate metadata.

    Args:
        raster_folder: Path to the folder containing raster files

    Yields:
        PixelUploadFile objects ready for upload
    """
    import re

    # Regular expression to match the pattern <some_name>_YYYY-MM-DD.tif
    pattern = re.compile(r"(.+)_(\d{4}-\d{2}-\d{2})\.tif$")

    # Find all .tif files in the folder
    for raster_file in raster_folder.glob("*.tif"):
        # Extract location and date from filename
        match = pattern.match(raster_file.name)
        if match:
            location = match.group(1)  # The part before the date
            date_str = match.group(2)  # The YYYY-MM-DD part

            try:
                # Parse the date from the filename
                capture_date = datetime.datetime.strptime(date_str, "%Y-%m-%d")

                # Find all support files that match the base name
                # (e.g., for Oslo_2022-09-07.tif, find Oslo_2022-09-07.tfw, Oslo_2022-09-07.tif.aux.xml, etc.)
                base_name = raster_file.stem  # Filename without extension
                support_files = [
                    f
                    for f in raster_folder.iterdir()
                    if f.is_file()
                    and f != raster_file
                    and (f.stem == base_name or f.name.startswith(f"{base_name}."))
                ]

                # Create and yield a PixelUploadFile with metadata
                yield PixelUploadFile(
                    file=raster_file,
                    support_files=support_files,
                    metadata=MetadataObject(
                        fields=RasterMetadataFields(
                            name=base_name,  # Use filename without extension as name
                            capture_date=capture_date,
                        ),
                        json_metadata={
                            "location": location,
                            "source": "DTM Survey",
                            "resolution": "1m",
                        },
                    ),
                )
            except ValueError:
                # Skip files with invalid date format
                print(f"Skipping file with invalid date format: {raster_file.name}")
                continue


def get_or_create_raster_collection(
    client: PixelClient,
    project_id: int,
    data_collection_name: str,
    data_collection_type: Literal["RGB", "DTM", "DSM"],
) -> dict[str, Any]:
    """
    Find an existing raster data collection or create a new one.

    Args:
        client: The Pixel client
        project_id: ID of the project to work with
        data_collection_name: Name of the data collection to use or create
        data_collection_type: Type of data collection (RGB, DTM, DSM)

    Returns:
        The data collection dictionary
    """
    if data_collection_type not in SUPPORTED_COLLECTION_TYPES:
        raise ValueError(
            f"Unsupported data collection type: {data_collection_type}. "
            f"Supported types are: {', '.join(SUPPORTED_COLLECTION_TYPES)}"
        )

    # Check if a data collection with the specified name already exists in the project
    data_collections = client.list_data_collections(
        project_id=project_id,
        data_collection_type=data_collection_type,
        name=data_collection_name,
    )

    if data_collections:
        # Use the existing data collection with the specified name
        data_collection = data_collections[0]
        print(
            f"Using existing data collection: {data_collection['name']} (ID: {data_collection['id']})"
        )

        # Verify that it's the correct type of data collection
        if data_collection["data_collection_type"] != data_collection_type:
            print(
                f"Warning: Existing data collection is not of type {data_collection_type}, "
                f"it's {data_collection['data_collection_type']}"
            )
            raise ValueError(
                f"Data collection type mismatch: expected {data_collection_type}, "
                f"got {data_collection['data_collection_type']}"
            )
        else:
            return data_collection

    # Configure raster info based on data collection type
    tags = []
    description = ""
    raster_info_args: dict[str, Any] = {
        "format": "GTiff",
    }

    if data_collection_type == "RGB":
        description = "RGB Imagery collection"
        tags = ["rgb", "imagery", "raster"]
        raster_info_args.update(
            {
                "data_type": "uint8",
                "num_bands": 3,
            }
        )
    elif data_collection_type == "DTM":
        description = "Digital Terrain Model collection"
        tags = ["dtm", "terrain", "elevation"]
        raster_info_args.update(
            {
                "data_type": "float32",
                "num_bands": 1,
            }
        )
    elif data_collection_type == "DSM":
        description = "Digital Surface Model collection"
        tags = ["dsm", "surface", "elevation"]
        raster_info_args.update(
            {
                "data_type": "float32",
                "num_bands": 1,
            }
        )

    # Create a new data collection with appropriate raster info
    data_collection = client.create_data_collection(
        project_id=project_id,
        name=data_collection_name,
        description=description,
        data_collection_type=data_collection_type,
        tags=tags,
        raster_info=RasterInfo.model_validate(raster_info_args),
    )
    print(
        f"Created new {data_collection_type} data collection: {data_collection['name']} "
        f"(ID: {data_collection['id']})"
    )
    return data_collection


def create_or_update_arcgis_service(
    client: PixelClient, service_name: str, data_collection: dict[str, Any]
) -> dict[str, Any]:
    """
    Create a new ArcGIS Image Service or update an existing one.

    Args:
        client: The Pixel client
        service_name: Name for the ArcGIS service
        data_collection: The data collection to include in the service

    Returns:
        The ArcGIS service dictionary
    """
    # Check if an ArcGIS Image Service exists with the specified name
    existing_services = client.list_arcgis_services(
        service_type="Image",
        name=service_name,
    )

    if existing_services:
        # Update the existing service
        service = existing_services[0]
        print(
            f"Found existing ArcGIS Image Service: {service['name']} (ID: {service['id']})"
        )

        # Check if our data collection is already included in the service
        current_data_collection_ids = [
            dc["id"] for dc in service.get("data_collections", [])
        ]
        if data_collection["id"] not in current_data_collection_ids:
            print(f"Adding data collection {data_collection['name']} to the service")

            # Update the service to include our data collection
            service = client.update_arcgis_service(
                service_type="Image",
                service_id=service["id"],
                update_input=ArcgisServiceUpdateInput(
                    data_collection_ids=current_data_collection_ids
                    + [data_collection["id"]],
                ),
            )
            print("Updated service with new data collection")

        # Refresh the service to include the new data
        service = client.refresh_arcgis_service(
            service_type="Image",
            service_id=service["id"],
            refresh_data=True,  # Refresh the data used by the service
            wait=True,  # Wait for the refresh to complete
        )["arcgis_service"]
    else:
        # Create a new ArcGIS Image Service
        print(f"Creating new ArcGIS Image Service: {service_name}")

        service = client.create_arcgis_service(
            service_type="Image",
            create_input=ArcgisServiceCreateInput(
                name=service_name,
                description=f"ArcGIS Image Service for {data_collection['data_collection_type']} data",
                data_collection_ids=[data_collection["id"]],
                options=ArcgisServiceCreateOptions(
                    image_service_options=ArcGISImageServiceOptions(
                        default_service_srid=25832,  # EPSG code for the service
                        wms_options=ArcGISImageServiceWMSOptions(
                            enable=True,  # Enable WMS capabilities
                            supported_srids=[
                                25832,
                                4326,
                            ],  # Supported coordinate systems
                            title=f"{service_name} WMS",
                            abstract=f"WMS service for {data_collection['data_collection_type']} data",
                        ),
                    ),
                ),
            ),
        )
        print(f"Created ArcGIS Image Service with ID: {service['id']}")

        # Start the service and wait for it to be ready
        service = client.start_arcgis_service(
            service_type="Image",
            service_id=service["id"],
            wait=True,  # Wait for the service to start
        )["arcgis_service"]
        print(f"Started ArcGIS Image Service: {service['name']}")

    # Print the service URL
    print(f"ArcGIS Image Service URL: {service['url']}")
    return service


def upload_and_optimize_rasters(
    client: PixelClient,
    project_id: int,
    data_collection_id: int,
    raster_folder: Path,
    data_collection_type: Literal["RGB", "DTM", "DSM"],
) -> list[dict[str, Any]]:
    """
    Upload raster files from a folder and optimize them with the appropriate profile.

    Args:
        client: The Pixel client
        project_id: ID of the project to work with
        data_collection_id: ID of the data collection to upload to
        raster_folder: Path to the folder containing raster files
        data_collection_type: Type of data collection (RGB, DTM, DSM)

    Returns:
        List of optimized rasters
    """
    # Upload raster files
    raster_files = list(iter_raster_files(raster_folder))
    print(f"Found {len(raster_files)} raster files to upload")

    if not raster_files:
        print("No raster files found.")
        return []

    rasters, errors = client.upload_multiple_rasters(
        project_id=project_id,
        data_collection_id=data_collection_id,
        files=raster_files,
        multipart=True,  # Use multipart upload for large files
        multipart_part_size=MULTIPART_PART_SIZE,
    )

    if errors:
        print("Errors during upload:")
        for error in errors:
            print(f"Job with id {error.job_id} failed with error: {error.detail}")

    print(f"Successfully uploaded {len(rasters)} rasters")

    # Optimize the rasters with the appropriate profile
    if not rasters:
        return []

    # Select the appropriate optimization profile based on data collection type
    profile = RGB_PROFILE if data_collection_type == "RGB" else TERRAIN_PROFILE

    optimized_rasters = client.optimize_rasters(
        project_id=project_id,
        data_collection_id=data_collection_id,
        profile=profile,
        raster_ids=[
            r["id"] for r in rasters
        ],  # Optimize only the newly uploaded rasters
    )

    print(f"Optimized {len(optimized_rasters)} rasters with {profile} profile")
    return optimized_rasters


def upload_optimize_deploy_workflow(
    project_id: int,
    raster_folder: Path,
    data_collection_name: str,
    data_collection_type: Literal["RGB", "DTM", "DSM"],
    service_name: str,
) -> None:
    """
    Complete workflow to:
    1. Create a raster data collection if it doesn't exist
    2. Upload raster files from a folder
    3. Optimize the rasters with the appropriate profile
    4. Create or update an ArcGIS Image Service

    Args:
        project_id: ID of the project to work with
        raster_folder: Path to the folder containing raster files
        data_collection_name: Name of the data collection to use or create
        data_collection_type: Type of data collection (RGB, DTM, DSM)
        service_name: Name for the ArcGIS service
    """
    client = get_client()

    # Step 1: Get or create the raster data collection
    data_collection = get_or_create_raster_collection(
        client, project_id, data_collection_name, data_collection_type
    )

    # Step 2 & 3: Upload and optimize raster files
    optimized_rasters = upload_and_optimize_rasters(
        client, project_id, data_collection["id"], raster_folder, data_collection_type
    )

    if not optimized_rasters:
        print("No rasters were optimized. Skipping ArcGIS service creation/update.")
        return

    # Step 4: Create or update an ArcGIS Image Service
    create_or_update_arcgis_service(client, service_name, data_collection)


if __name__ == "__main__":
    import argparse
    import sys

    def validate_project_id(value):
        """Validate that project_id is a positive integer."""
        try:
            ivalue = int(value)
            if ivalue <= 0:
                raise argparse.ArgumentTypeError(
                    f"Project ID must be a positive integer, got {value}"
                )
            return ivalue
        except ValueError:
            raise argparse.ArgumentTypeError(
                f"Project ID must be an integer, got {value}"
            )

    def validate_folder_path(value):
        """Validate that the folder path exists and is a directory."""
        path = Path(value)
        if not path.exists():
            raise argparse.ArgumentTypeError(f"Folder path does not exist: {value}")
        if not path.is_dir():
            raise argparse.ArgumentTypeError(f"Path is not a directory: {value}")
        return path

    def validate_name(value):
        """Validate that the name is not empty and has a reasonable length."""
        if not value or not value.strip():
            raise argparse.ArgumentTypeError("Name cannot be empty")
        if len(value) > 100:
            raise argparse.ArgumentTypeError(
                f"Name is too long (max 100 characters): {value}"
            )
        return value

    parser = argparse.ArgumentParser(
        description="Upload rasters, optimize them, and deploy to ArcGIS service"
    )
    parser.add_argument(
        "--project-id",
        type=validate_project_id,
        required=True,
        help="ID of the project to work with",
    )
    parser.add_argument(
        "--raster-folder",
        type=validate_folder_path,
        required=True,
        help="Path to the folder containing raster files",
    )
    parser.add_argument(
        "--data-collection-name",
        type=validate_name,
        required=True,
        help="Name of the data collection to use or create",
    )
    parser.add_argument(
        "--data-collection-type",
        type=str,
        choices=SUPPORTED_COLLECTION_TYPES,
        required=True,
        help=f"Type of data collection: {', '.join(SUPPORTED_COLLECTION_TYPES)}",
    )
    parser.add_argument(
        "--service-name",
        type=validate_name,
        required=True,
        help="Name for the ArcGIS service",
    )

    args = parser.parse_args()

    try:
        upload_optimize_deploy_workflow(
            project_id=args.project_id,
            raster_folder=args.raster_folder,  # Already a Path object from validation
            data_collection_name=args.data_collection_name,
            data_collection_type=args.data_collection_type,
            service_name=args.service_name,
        )
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

File Naming Convention¶

The script expects raster files with the naming pattern <some_name>_YYYY-MM-DD.tif and their associated support files in the same directory. For example:

raster_folder/
├── Oslo_2022-09-07.tif
├── Oslo_2022-09-07.tfw
├── Oslo_2022-09-07.tif.aux.xml
├── Oslo_2023-10-27.tif
├── Oslo_2023-10-27.tfw
├── Oslo_2023-10-27.tif.aux.xml
├── Oslo_2024-09-13.tif
├── Oslo_2024-09-13.tfw
└── Oslo_2024-09-13.tif.aux.xml

The script will automatically:

Extract the location name (e.g., "Oslo") from the filename
Extract the date (e.g., "2022-09-07") from the filename
Find all support files that match the base name of each raster file
Include the extracted information in the metadata when uploading

Running the Script¶

You can run the script from the command line with the following parameters:

Warning

Make sure to set the required environment variables for authentication as described in the Authentication section. You can change the script to use a .env with PixelApiSettings.from_env_file("path/to/.env") to load the settings from a file. (see PixelApiSettings class).

python pixel_raster_workflow.py \
  --project-id 123 \
  --raster-folder /path/to/raster_folder \
  --data-collection-name "Oslo DTM Collection" \
  --data-collection-type DTM \
  --service-name "DTM_Service"

Available data collection types are:

RGB: For RGB imagery (optimized with RGB profile)
DTM: For Digital Terrain Models (optimized with terrain profile)
DSM: For Digital Surface Models (optimized with terrain profile)

The script will:

Check if a data collection with the specified name and type exists in the project
Create a new data collection if needed, with appropriate settings for the specified type
Upload all raster files found in the folder structure
Optimize the rasters with the appropriate profile
Check if an ArcGIS Image Service with the specified name exists
Update the existing service or create a new one
Ensure the data collection is included in the service
Refresh the service to include the new data

This workflow is particularly useful for automating the process of updating raster data in ArcGIS services, especially when dealing with time-series data that needs to be regularly updated.