Skip to content

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.

Access Management

These examples cover common access tasks. Most of them need a signed-in user rather than an api key; see Access management for the concepts and for which calls need one.

Onboarding a team

Sets up access for a new team: creates a group, creates the team's users in it, gives the group the editor role on two projects, creates an api key for the team's upload pipeline that follows the group, and checks what the group can reach. Run it as a superuser.

from pixel_client import get_client

# Run this as a superuser signed in with a username and password: creating
# users needs a superuser, and api keys cannot be created with an api key.
client = get_client()

# 1. Create a group for the team. You become its first member and admin.
group = client.create_group("Survey team", "Field survey crew, 2026")
group_id = group["id"]
print(f"Created group {group['name']} with ID {group_id}")

# 2. Create the team's users straight into the group. Each password is
#    temporary: the user chooses their own the first time they sign in.
team = [
    ("kari.nordmann@example.com", "Kari", "Nordmann"),
    ("ola.nordmann@example.com", "Ola", "Nordmann"),
]
for email, first_name, last_name in team:
    user = client.create_user(
        email=email,
        first_name=first_name,
        last_name=last_name,
        password="<temporary password>",
        groups=[group_id],
    )
    print(f"Created user {user['email']}")

# 3. Give the group the editor role on two projects. Every member gets it, on
#    the projects and their subprojects.
for change in client.grant_project_access([42, 43], "editor", groups=[group_id]):
    print(f"Project {change['resource']['id']}: {change['action']}")

# 4. Create an api key for the team's upload pipeline. The key follows the
#    group, so it reaches whatever the group can reach, also after the group's
#    projects change. It is still limited to your own access.
key = client.create_api_key(
    "Survey upload pipeline", groups=[group_id], expires_in_days=180
)
print(f"Api key {key['api_key']['id']} expires {key['api_key']['expires_at']}")
# The token is shown only this once. Store it in the pipeline's secrets as
# PIXEL_API_KEY.
pipeline_token = key["token"]

# 5. Check what the group can do now.
access = client.get_group_access(group_id)
print(f"{group['name']} has {access['member_count']} members")
for project in access["projects"]:
    print(f"- {project['project_name']}: {project['role']}")

Offboarding a person

Removes a person's access when they leave: reads why they have access with get_user_access, revokes the roles they hold directly, takes them out of the groups that gave them access, and revokes their api keys. Run it as a superuser. Deactivating the account itself is done in the web application.

from pixel_client import get_client

# Run this as a superuser signed in with a username and password: looking up
# another user's access and their api keys needs it, and api keys cannot be
# used for either.
client = get_client()

email = "kari.nordmann@example.com"  # The person who is leaving

# 1. See what the person can reach, and why. Every project lists the grants
#    behind the person's access there: given to them directly, or through a
#    group they are a member of.
access = client.get_user_access(email)
direct_projects: set[int] = set()
group_ids: dict[int, str] = {}
for project in access["projects"]:
    for grant in project["grants"]:
        source = grant["source"]
        if source["kind"] == "group":
            group_ids[source["group_id"]] = source["group_name"]
        elif grant["held_on"] == "project":
            # Held directly on `project_id`, which can be a parent project
            direct_projects.add(grant["project_id"])

# 2. Remove the roles they hold directly on projects.
if direct_projects:
    for change in client.revoke_project_access(sorted(direct_projects), users=[email]):
        print(f"Project {change['resource']['id']}: {change['action']}")

# 3. Take them out of the groups that gave them access. A group must keep at
#    least one admin, so this fails if they are the last admin of a group.
for group_id, group_name in group_ids.items():
    client.remove_group_member(group_id, email)
    print(f"Removed from group {group_name}")

# 4. Revoke the api keys they own, so their scripts stop working. Keys stop
#    working anyway once their owner is deactivated, but revoking is explicit.
for key in client.list_api_keys(all=True):
    if key["owner"]["email"] == email and key["status"] in ("active", "owner_inactive"):
        client.revoke_api_key(key["id"])
        print(f"Revoked api key {key['name']}")

# 5. Check the result. Access from the person's account type (shown under
#    `root`) is unchanged; change that with `update_user(email, user_type=...)`.
#    Deactivating the account is done in the web application.
remaining = client.get_user_access(email)["projects"]
print(f"{email} can still reach {len(remaining)} project(s)")

Reporting who has access

Exports every user and group grant on the projects you administer to a CSV file, with one row per person and project and whether the access is direct, inherited, or through a group. Grants show what was given; get_user_access shows what someone can actually do once all sources are combined.

import csv

from pixel_client import get_client

# Run this signed in with a username and password: the grant listing does not
# accept api keys. It covers the projects you administer.
client = get_client()

# 1. Fetch every user and group grant on the projects you administer. Each
#    grant is listed once, on the project where it applies.
grants = client.list_grants()

# 2. Group grants list their members' user IDs. Look up the names once, so the
#    report can show who gets access through each group.
member_ids = sorted({uid for g in grants for uid in g["member_ids"] or []})
names = {
    user["id"]: f"{user['first_name']} {user['last_name']}"
    for user in (client.list_users(ids=member_ids) if member_ids else [])
}

# 3. Write one row per person and project, marking how they got access.
with open("access_report.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["project", "person or group", "role", "how"])
    for grant in grants:
        # "project": held on this project; "above": inherited from a project
        # above the ones you administer, which you cannot change here.
        how = "direct" if grant["held_on"] == "project" else "inherited from above"
        if grant["principal_kind"] == "user":
            writer.writerow(
                [grant["project_name"], grant["principal_name"], grant["role"], how]
            )
            continue
        writer.writerow(
            [grant["project_name"], f"group {grant['principal_name']}", grant["role"], how]
        )
        for user_id in grant["member_ids"] or []:
            writer.writerow(
                [
                    grant["project_name"],
                    names.get(user_id, f"user {user_id}"),
                    grant["role"],
                    f"via group {grant['principal_name']}",
                ]
            )

print(f"Wrote {len(grants)} grants to access_report.csv")

# 4. Grants only show what was given. To see what one person can actually do
#    on each project, with access from every source combined, use
#    get_user_access.
me = client.me()
for project in client.get_user_access(me["id"])["projects"]:
    print(f"You are {project['role']} on {project['project_name']}")

Rotating api keys before they expire

Finds your active api keys that expire within two weeks and replaces each with a new key that has the same scope. The new token is returned only once, so the example hands it to a function where you store it in your secret store. It also warns about keys that have lost access because you have.

from datetime import datetime, timedelta, timezone

from pixel_client import get_client

# Run this signed in with a username and password: api keys cannot manage
# api keys.
client = get_client()

ROTATE_WITHIN = timedelta(days=14)
now = datetime.now(timezone.utc)


def store_secret(key_name: str, token: str) -> None:
    """Replace this with your secret store: Azure Key Vault, AWS Secrets
    Manager, your CI system's secret variables, ..."""
    print(f"Store the new token for {key_name!r} now (ends in ...{token[-4:]})")


for key in client.list_api_keys():
    # 1. Warn about keys that can do less than they were given: their owner
    #    (you) has lost some of the access the key was scoped to.
    if key["weakened"]:
        print(f"Key {key['name']!r} has lost access; check its grants")

    # 2. Replace active keys that expire soon. The new key gets the same scope,
    #    and the old key is revoked as soon as the new one exists, so update
    #    the secret right away. (An expired key can be renewed the same way,
    #    but it stays in the list as expired, so pick those by hand.)
    expires_at = datetime.fromisoformat(key["expires_at"])
    if key["status"] != "active" or expires_at - now >= ROTATE_WITHIN:
        continue
    new_key = client.replace_api_key(
        key["id"], name=key["name"], expires_in_days=90
    )
    # The token is returned only this once.
    store_secret(key["name"], new_key["token"])
    print(
        f"Replaced {key['name']!r}; new key expires {new_key['api_key']['expires_at']}"
    )

# 3. Replacing is also how you change a key's scope: each of projects, groups
#    and base_role that you pass replaces that part of the old scope.
# client.replace_api_key("<key id>", projects={42: "viewer"})

Sharing a map service with ArcGIS users

Gives an external partner access to one ArcGIS feature service: creates a geo service user with a generated password, grants it access to the service, and shows how to pause the access by disabling the user. A service anyone may see can be made public instead. Run it as a superuser; geo service users work on ArcGIS Server v2.

from pixel_client import get_client
from pixel_client.models import ListParams

# Run this as a superuser signed in with a username and password. Geo service
# users work on ArcGIS Server v2 only.
client = get_client()

# 1. Find the map service to share.
services = client.list_arcgis_services(
    "Feature", params=ListParams(name="Bridge inspections 2026")
)
service = services[0]
print(f"Sharing {service['name']} ({service['url']})")

# 2. Create a geo service user for the partner. Leave out the password to get
#    a strong generated one. It is returned only this once, so hand it over
#    securely now. The username cannot be changed later.
created = client.create_geo_service_user(
    "partner-bridgeco", description="BridgeCo inspection team, ArcGIS Pro"
)
print(f"Username: {created['geo_service_user']['username']}")
print(f"Password: {created['password']}")

# 3. Give the geo service user access to the service. Geo service users are
#    given by ID or username. Changes reach the map service within minutes.
client.grant_arcgis_service_access("Feature", service["id"], ["partner-bridgeco"])
for user in client.list_arcgis_service_access("Feature", service["id"]):
    print(f"{user['username']} has access")

# 4. When the partner should lose access for a while, disable the user: its
#    grants are kept and come back with enable_geo_service_user. When the
#    collaboration ends, delete the user instead.
client.disable_geo_service_user("partner-bridgeco")
# client.enable_geo_service_user("partner-bridgeco")
# client.delete_geo_service_user("partner-bridgeco")

# 5. A service everyone may see does not need geo service users at all: make
#    it public instead. Services are private by default.
# from pixel_client.models import ArcgisServiceUpdateInput
# client.update_arcgis_service(
#     "Feature", service["id"], ArcgisServiceUpdateInput(is_public=True)
# )

Giving a group access to every project with a tag (async)

Gives a group the viewer role on every project with a tag, using the asynchronous client. Projects where the group already has a role are skipped, because a grant sets the role and would lower a higher one. The script can run on a schedule and only picks up new projects; failures on single projects are reported without stopping the rest.

import asyncio

from pixel_client import get_client
from pixel_client.exceptions import PixelAccessUpdateError
from pixel_client.models import ListParams


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

    group_id = 3  # The group that should see every project tagged "bridges"

    # 1. Find the projects with the tag, and the projects the group already
    #    has a role on. Independent requests can run concurrently.
    projects, group = await asyncio.gather(
        client.list_projects(params=ListParams(tags=["bridges"], limit=500)),
        client.get_group(group_id),
    )
    has_role = {project["id"] for project in group["projects"]}
    project_ids = [p["id"] for p in projects if p["id"] not in has_role]
    print(f"{len(project_ids)} of {len(projects)} projects tagged 'bridges' are new")
    if not project_ids:
        return

    # 2. Give the group the viewer role on the new ones in one call. A grant
    #    sets the role, so projects where the group already has a role were
    #    left out above: granting viewer there would lower an editor or admin
    #    role. The script can run every night and only picks up new projects.
    try:
        changes = await client.grant_project_access(
            project_ids, "viewer", groups=[group_id]
        )
    except PixelAccessUpdateError as e:
        # The other projects were still granted. Typical cause: you are not an
        # admin on some of the projects.
        changes = e.succeeded
        for change, error in e.failed:
            print(f"Could not grant on project {change['resource']['id']}: {error}")
    print(f"Granted on {[c['resource']['id'] for c in changes]}")

    # 3. Look up what the group can reach now, subprojects included.
    access = await client.get_group_access(group_id)
    print(f"{group['name']} reaches {len(access['projects'])} projects")


asyncio.run(main())

Analysis

Pausing and resuming analysis tasks

Pauses every running task of an analysis service, for example before planned maintenance of a system the analysis depends on, waits until they are all PAUSED, and resumes them afterwards. Only modules marked pausable allow it, and a task always finishes its current step before it pauses.

import time

from pixel_client import get_client
from pixel_client.models import AnalysisTaskStatusEnum

client = get_client()

service_id = 12  # The analysis service whose tasks should be paused

# 1. Only modules marked `pausable` let you pause their tasks.
service = client.list_analysis_services(ids=[service_id])[0]
module = next(
    m for m in client.list_analysis_modules() if m["name"] == service["module_name"]
)
if not module["pausable"]:
    raise SystemExit(f"Tasks of the {module['name']} module cannot be paused")

# 2. Pause every running task, for example before planned maintenance of a
#    system the analysis depends on. A task finishes its current step first:
#    it is PAUSING until then, and PAUSED once nothing is running.
running = client.list_analysis_tasks(
    service_id, status=AnalysisTaskStatusEnum.RUNNING, limit=1000
)
paused_ids = [client.pause_analysis_task(service_id, task["id"])["id"] for task in running]
print(f"Pausing {len(paused_ids)} task(s)")

while paused_ids:
    tasks = client.list_analysis_tasks(service_id, ids=paused_ids, limit=len(paused_ids))
    if all(task["status"] == AnalysisTaskStatusEnum.PAUSED for task in tasks):
        break
    time.sleep(10)
print("All tasks paused")

# 3. Later, resume them. Each continues from the step it stopped at. Resuming a
#    task that is still PAUSING cancels the pause instead.
for task_id in paused_ids:
    result = client.resume_analysis_task(service_id, task_id)
    print(f"Task {result['id']}: {result['status']}")

Complex Workflows

Raster Upload, Optimization, and ArcGIS Service Deployment

This example demonstrates a complete workflow for:

  1. Finding or creating a raster data collection (RGB, DTM, or DSM) if it doesn't exist
  2. Uploading raster files from a folder structure
  3. Optimizing the rasters with the appropriate profile (RGB or terrain)
  4. 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:

  1. Extract the location name (e.g., "Oslo") from the filename
  2. Extract the date (e.g., "2022-09-07") from the filename
  3. Find all support files that match the base name of each raster file
  4. 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:

  1. Check if a data collection with the specified name and type exists in the project
  2. Create a new data collection if needed, with appropriate settings for the specified type
  3. Upload all raster files found in the folder structure
  4. Optimize the rasters with the appropriate profile
  5. Check if an ArcGIS Image Service with the specified name exists
  6. Update the existing service or create a new one
  7. Ensure the data collection is included in the service
  8. 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.