HASTE Queue Functions

HASTE Queue Functions#

Azure Functions backend for asynchronous, queue-driven processing in HASTE. These functions act as orchestrators — they receive a queue message, submit a job to Azure Batch for the actual compute, and poll for completion across subsequent invocations by re-queuing themselves.

All functions are defined in function_app.py.

Architecture#

HTTP API → Azure Storage Queue → Azure Function (trigger)
                                        │
                                        ▼
                               Submit task to Azure Batch
                                        │
                                        ▼
                            Azure Batch Pool (GPU VMs)
                            runs Docker container task
                                        │
                               ┌────────┴────────┐
                               │  IN_PROGRESS?   │
                               │  Enqueue next   │← function exits after sending a new message
                               └────────┬────────┘  (delay via send_message visibility_timeout)
                                        │
                                   COMPLETED / FAILED
                                        │
                                  Save metadata,
                                  queue artifact zip

Each function invocation either starts a new Batch task or checks the status of an existing one. When a task is still running, the function re-queues itself and exits — the Azure Functions runtime picks up the next message and checks again. This means a single training or inference job can span many function invocations over hours.

Azure Batch Pools#

There are two dedicated pools:

Pool	Env Var	Default ID	Used By
Imagery prep	`AZURE_BATCH_IMAGERYPREP_POOL_ID`	`imageryprep-pool`	`GetProcessImageLayerQueueMessage`
Training / Inference	`AZURE_BATCH_TRAINING_POOL_ID`	`training-pool`	`GetCreateModelRunQueueMessage`, `GetRunInferenceQueueMessage`

Each pool runs tasks as Docker containers pulled from Azure Container Registry using managed identity. Containers run with --shm-size=32g for GPU workloads.

Queue Triggers#

GetProcessImageLayerQueueMessage#

Queue: image_queue_name | Pool: imageryprep-pool

Preprocesses a newly uploaded geospatial image layer. On each invocation:

PENDING → submits a Batch task via ImageryPostProcessor; task runs prepare-imagery CLI inside the container
IN_PROGRESS → reads imagery_friendly.log from the task working directory for progress, then re-queues
COMPLETED → reads imagery_manifest.json for output paths (mosaics, COGs, building footprints, valid-area mask); generates label project files, converts to GeoJSON, stores artifacts
FAILED → saves the layer with a FAILED status and error message

If the layer was deleted before processing starts, the message is silently skipped.

GetCreateModelRunQueueMessage#

Queue: train_queue_name | Pool: training-pool

Orchestrates ML model training. On each invocation:

PENDING → submits a Batch training task; input files include labels (GeoJSON), pre/post-event COG imagery, optional initial weights, and experiment config (YAML)
IN_PROGRESS → parses TensorBoard event files from the task working directory to extract per-epoch metrics (completed epochs, time per epoch, ETA), then re-queues
COMPLETED → extracts final metrics; if autoRunInference is set, automatically enqueues an inference job
CANCELLED → calls TrainPostprocessor.cancel() which terminates the Batch task
Post-run (separate error scope) → queues artifact zipping for FAILED/CANCELLED training runs when trainingOutputPath exists (completed runs can be zipped on-demand via PutArtifactsZipQueueMessage)

GetRunInferenceQueueMessage#

Queue: inference_queue_name | Pool: training-pool

Runs model inference on a geospatial image layer. On each invocation:

PENDING → submits a Batch inference task; inputs include COG imagery, building footprints GeoPackage, training checkpoint (last.ckpt), and experiment config
IN_PROGRESS → reads workflow_progress.log from the task for user-visible progress messages, then re-queues
COMPLETED → surfaces output artifact URLs: predicted damage layer (GeoTIFF) and vector predictions (GeoPackage)
CANCELLED → calls InferencePostprocessor.cancel() which terminates the Batch task
Post-run (separate error scope) → queues artifact zipping if output path exists

Note: Raw stderr.txt from the Batch task is never returned to the client — only sanitized workflow_progress.log messages are surfaced.

UpdateStatsMessage#

Queue: stats_queue_name

Lightweight function — no Batch involvement. Deserializes a StatsRequest message and runs StatsPostProcessor to recalculate and persist aggregated ProjectsSummary metadata.

GetArtifactsZipQueueMessage#

Queue: zip_queue_name

Packages training/inference output files into a downloadable zip archive via ArtifactProcessor.process_zip(). Triggered automatically by the training and inference functions after their runs complete (or fail with partial output).

ImagePoisonQueueHandler#

Queue: {image_queue_name}-poison (dead-letter)

Handles image layer messages that exceeded the max dequeue count (maxDequeueCount=1 in host.json). Marks the image layer as FAILED in metadata so the UI reflects the error. Training and inference failures are caught inline and do not have a poison handler.

Error Handling#

Poison queue: Failed image processing messages are routed to the -poison dead-letter queue by the Azure Functions runtime after 1 failed attempt.
Tiered try/catch: Training and inference use separate error scopes for each phase (main processing, artifact zipping, inference trigger) so a failure in one phase does not block the others.
Status persistence: All failure paths attempt to write a FAILED or CANCELLED status back to metadata, even when the primary operation throws.
Batch retry: The AzureBatchRunner uses exponential backoff (4–10 s, up to 5 attempts) via tenacity for transient Azure Batch API errors (5xx).

Configuration#

Azure Batch#

Environment Variable	Description	Default
`AZURE_BATCH_ACCOUNT_NAME`	Batch account name	—
`AZURE_BATCH_ACCOUNT_KEY`	Batch account key (if not using managed identity)	—
`AZURE_BATCH_ACCOUNT_URL`	Batch account endpoint URL	—
`AZURE_BATCH_TRAINING_POOL_ID`	Pool for training and inference	`training-pool`
`AZURE_BATCH_IMAGERYPREP_POOL_ID`	Pool for imagery preprocessing	`imageryprep-pool`
`AZURE_BATCH_TARGET_DEDICATED_NODES`	Number of dedicated nodes per pool	`1`
`AZURE_BATCH_TARGET_LOW_PRIORITY_NODES`	Number of spot nodes per pool	`0`
`AZURE_BATCH_TASK_RETENTION_TIME`	ISO 8601 retention for task files	`P2D`
`TRAINING_BATCH_JOB_ID`	Job ID for training tasks	Pool ID
`INFERENCE_BATCH_JOB_ID`	Job ID for inference tasks	Pool ID
`IMAGERYPREP_BATCH_JOB_ID`	Job ID for imagery prep tasks	Pool ID
`ARTIFACT_BATCH_JOB_ID`	Job ID for artifact zip tasks	Pool ID

Container Registry#

Environment Variable	Description
`AZURE_BATCH_REGISTRY_SERVER`	ACR login server (e.g. `myregistry.azurecr.io`)
`AZURE_BATCH_DOCKER_IMAGE`	Training/inference Docker image (full tag)
`AZURE_BATCH_IMAGERYPREP_DOCKER_IMAGE`	Imagery preprocessing Docker image (full tag)
`AZURE_BATCH_REGISTRY_IDENTITY_RESOURCE_ID`	Managed identity resource ID for ACR pull

Queue & Storage#

Environment Variable	Description
`AzureWebJobsStorage`	Azure Storage connection string (queues + blobs)
`DATA_DIR`	Root directory for app data and per-process log files

Deployment#

The app ships as a Docker container.

Base image: mcr.microsoft.com/azure-functions/python:4-python3.11

The container runs as a non-root appuser. Local hastegeo source is copied in alongside the installed wheel to support development builds. Set HASTE_SKIP_VERSION_BUMP=1 during the build to prevent version bumping.

docker build -t hastefuncqueues .

host.json concurrency settings:

batchSize=1 — process one queue message at a time per worker (one heavy job at a time)
maxDequeueCount=1 — immediately dead-letter on failure
Function timeout: 23:59:59 — supports long-running training and inference jobs

Development Setup#

Prerequisites#

Install Azure Functions Core Tools:

npm install -g azure-functions-core-tools@4 --unsafe-perm true

Running Locally#

Set AzureWebJobsStorage and other required variables in local.settings.json, then:

func start

Or use the Launch Functions VS Code launch configuration.

Local Debugging#

Add a breakpoint in code:

breakpoint()

Then use the Launch Functions VS Code task. The terminal drops into a pdb prompt when execution reaches the breakpoint.

Note: The VS Code visual breakpoint UI will not work — the Azure Functions process is not attached to the VS Code Python debugger.

Auto-generated API Docs#

DeleteModelCatalog(req) [source]

Delete a model from the model catalog.

This endpoint removes a catalogued model from the model catalog. The model

can be identified by either its baseModelName or modelId.

Returns:: func.HttpResponse: JSON response containing: - success (bool): Operation success status - message (str): Success or error message - deletedModel (dict): The deleted catalog model object

GenerateProjectStats(req) [source]

Helper endpoint to generate project stats from all project data.

Useful for reconciling if they are out of sync.

GetAssessmentReport(req) [source]

Run the damage-assessment evaluation against the model’s inference.

Reproduces the “Analyze results” computation from the

``notebooks/Evaluate-Gezanine_1.ipynb`` notebook and from the

``validation/evaluate.py`` CLI: precision/recall/AP against any

available human labels, the predicted damaged-building count for the

layer, and a finite-population estimate (with 95% CI) for the total

damaged count across all footprints above a minimum area threshold.

The math lives in :func:`hastegeo.core.utils.assessment.compute_assessment_report`.

This endpoint is the I/O wrapper that pulls the model’s merged

predictions GeoPackage and the layer’s cached building footprints

GeoPackage, joins them on row order (the convention used by

``merge_with_building_footprints.py``), and folds in whatever

validation labels exist.

Query params:

projectId (str): Parent project identifier.

imageLayerId (str): Image layer identifier.

modelId (str): Model whose inference results to assess.

threshold (float, optional): Damage fraction above which a

building is called damaged (default 0.1, same as the CLI).

minAreaM2 (float, optional): Minimum footprint area in m² for

the population extrapolation (default 50).

GetAzureMapsToken(req) [source]: Return an Azure AD token for Azure Maps using managed identity.

GetBuildingEmbeddingsGeoJSON(req) [source]

Return the full building-embeddings GeoJSON for the interactive labeler.

Loads the embedding Model’s ``embeddingsGeoJSONUrl`` (footprints + f_*

feature columns, one row per footprint in row-index order) and streams it

back. Unlike GetBuildingFootprintsGeoJSON this does NOT sample — the

in-browser model needs every building’s full feature vector.

Query params: projectId, imageLayerId, modelId.

GetBuildingFootprintsGeoJSON(req) [source]

Return a random sample of building footprints as a GeoJSON FeatureCollection.

Reads the cached .gpkg file from blob storage for the given image layer,

selects a random sample of up to ``sample`` buildings, and returns them as

a GeoJSON FeatureCollection. Each feature carries the Overture building

``id``, ``subtype``, and ``class`` properties.

Query params:

projectId (str): Parent project identifier.

imageLayerId (str): Image layer identifier.

sample (int, optional): Maximum number of buildings to return

(default 200). Clamped to the inclusive range [1, 2000] to bound

response size and server-side memory.

GetBuildingValidation(req) [source]

Return existing building validation labels for an image layer.

Query params:

projectId (str): Parent project identifier.

imageLayerId (str): Image layer identifier.

Returns a BuildingValidation JSON object, or an empty one if no labels exist yet.

GetCreateModelRunQueueMessage(msg) [source]

Execute machine learning model training workflows from queue messages.

This function orchestrates complete model training pipelines including:

- Training data preparation and validation

- Model architecture configuration and initialization

- Distributed training execution with progress monitoring

- Model validation and performance evaluation

- Artifact generation and storage (checkpoints, logs, metrics)

- Integration with Azure Machine Learning services

The training pipeline supports:

- Multiple model architectures (CNN, transformer-based, etc.)

- Distributed training across multiple compute nodes

- Automatic hyperparameter optimization

- Real-time progress tracking and logging

- Checkpoint saving for resumable training

- Model performance validation and testing

Returns:: None: This is a queue trigger function with no return value. Training progress is tracked through Azure ML and metadata updates.

GetDashboardData(req) [source]

Retrieve comprehensive dashboard statistics for the HASTE application.

This endpoint aggregates project summary statistics including project counts,

layer information, model status, and system-wide metrics. If cached statistics

are not available, it will automatically generate them.

Returns:: func.HttpResponse: JSON response containing dashboard statistics with the following structure: - projects: List of project summaries sorted by creation date (newest first) - project_count: Total number of projects in the system - layer_count: Total number of image layers across all projects - model_count: Total number of models across all projects - Additional aggregated statistics

GetInteractiveLabels(req) [source]

Return the interactive labeler’s labels for an embedding model.

These live in a separate (model-scoped) store from the layer-scoped

Building Validation labels so the two workflows stay independent.

Query params: projectId, modelId. Returns {“labels”: {…}} (empty if none).

GetModelArtifact(req) [source]

Stream an embedding model’s browser artifact via managed identity.

The Interactive Labeler reads the PMTiles archive (and its features

sidecar) by HTTP byte-range straight from the browser. Handing the

client a direct ``*.blob.core.windows.net`` SAS URL only works from

IPs on the storage firewall allowlist, so remote/mobile/external

labelers get a 403. This route keeps the standard HASTE pattern: the

browser fetches same-origin ``/api`` and the function app does the

blob I/O server-side over the Azure backbone, honoring ``Range`` so

pmtiles.js can do partial reads.

Supported ``kind`` values: ``pmtiles``, ``sidecar`` and ``geojson``

(fetched/parsed in-browser), plus ``gpkg`` — the per-building

predictions GeoPackage saved by ``PutBuildingPredictions``, served as

a downloadable attachment. Example:

``GET /api/GetModelArtifact?projectId=<pid>&modelId=<mid>&kind=gpkg``.

GetModelCatalog(req) [source]

Retrieve the model catalog containing all available base models for training.

This endpoint returns a comprehensive list of all catalogued models that can be used

as base models for training new models. The catalog includes metadata about each model

such as disaster type, imagery source, and usage history.

Returns:: func.HttpResponse: JSON response containing: - modelCatalog (list): Array of catalogued models with metadata: - baseModelName (str): Human-readable name of the base model - modelId (str): Original model ID this was checkpointed from - projectId (str): Project ID where the model was created - imageLayerId (str): Image layer ID associated with the model - imagerySource (str): Source of imagery (Planet, Maxar, etc.) - eventTypes (list[str]): Types of disaster events (Hurricane, Tornado, Fires, etc.) - cataloguedDate (str): ISO timestamp when model was added to catalog - cataloguedByUser (str): User ID who added the model to catalog - additionalInfo (dict): User-defined metadata key-value pairs - usedByModels (list): List of model IDs that used this as base model

GetProcessImageLayerQueueMessage(msg) [source]

Process image layer queue messages for geospatial imagery preprocessing.

This function handles asynchronous processing of uploaded image layers, including:

- Geospatial transformation and projection corrections

- Image tiling and pyramid generation for efficient viewing

- Metadata extraction and validation

- Thumbnail generation for preview purposes

- Integration with storage systems and databases

The processing pipeline includes:

1. Message validation and deserialization

2. Existence check to prevent duplicate processing

3. Geospatial processing using GDAL operations

4. Tile generation for web mapping interfaces

5. Metadata storage and indexing

6. Status updates and progress tracking

Returns:: None: This is a queue trigger function with no return value. Status is tracked through logging and metadata updates.

GetProjectDetails(req) [source]

Retrieve comprehensive details for a specific HASTE project.

This endpoint returns detailed information about a project including metadata,

image layers, associated models, processing status, and configuration. It supports

optional inclusion of model details to reduce response size when not needed.

The response includes:

- Project metadata (name, description, creation date, location)

- Image layer information with processing status

- Model configurations and training status (if requested)

- Primary class definitions and labeling schema

- Processing statistics and progress indicators

Returns:: func.HttpResponse: JSON response containing: - project (dict): Complete project information including: - projectId (str): Unique project identifier - name (str): Human-readable project name - description (str): Project description - location (dict): Geographic location information - affectedCountries (list): Countries affected by the event - eventDate (str): ISO timestamp of the event - creationDate (str): ISO timestamp of project creation - imageLayerCount (int): Number of associated image layers - imageLayer (list): Detailed image layer information - primaryClasses (list): Classification schema definitions - models (list, optional): Model details if includeModels=true - processingStatus (dict): Current processing state

GetProjects(req) [source]

Retrieve all projects with their associated statistics and metadata.

This endpoint returns a comprehensive list of all projects in the HASTE system

along with aggregated statistics including layer counts, model counts, and

processing status information.

Returns:: func.HttpResponse: JSON response containing project statistics with the following structure: - projects: Array of project objects with full metadata - project_count: Total number of projects - layer_count: Total number of image layers across all projects - model_count: Total number of models across all projects

GetRunEmbeddingQueueMessage(msg) [source]

Execute a building-embedding job (building labeling workflow).

Deserializes the embedding Model, loads its image layer, and drives the

EmbeddingPostprocessor state machine (submit -> poll -> finalize). No

zip/inference follow-on. Re-enqueues itself while in progress.

GetRunInferenceQueueMessage(msg) [source]

Execute model inference on geospatial imagery for prediction generation.

This function handles batch inference processing using trained models to generate

predictions on new imagery data. The inference pipeline includes:

- Model loading and initialization from stored checkpoints

- Input imagery preprocessing and tile generation

- Batch prediction execution with GPU acceleration

- Post-processing of predictions (confidence scoring, filtering)

- Result aggregation and visualization preparation

- Integration with mapping and visualization services

Inference capabilities:

- Large-scale imagery processing with tiling strategies

- Multi-class and multi-label prediction support

- Confidence threshold filtering and uncertainty quantification

- Geospatial coordinate alignment and projection handling

- Output format conversion (raster, vector, tiles)

- Real-time progress monitoring and status updates

Returns:: None: This is a queue trigger function with no return value. Results are stored and status tracked through metadata updates.

GetValidationReport(req) [source]

Compute a validation accuracy report by crossing inference results with

user-supplied building validation labels.

Joins the inference GeoPackage (integer sequential IDs, ``damaged`` 0/1)

with the building-footprints GeoPackage (Overture string IDs in the same

row order) to produce a per-building prediction lookup, then compares

against the human labels stored in BuildingValidation.

``Unknown`` labels are excluded from all metric calculations.

Query params:

projectId (str): Parent project identifier.

imageLayerId (str): Image layer identifier.

modelId (str): Model identifier whose inference results to use.

Returns JSON:

{

“matched”: int, // buildings with both a prediction and a label

“totalValidationLabels”: int,

“labelCounts”: {“Damaged”: int, “NotDamaged”: int, “Unknown”: int},

“accuracy”: float,

“confusionMatrix”: {

“labels”: [“Damaged”, “NotDamaged”],

“matrix”: [[TP, FN], [FP, TN]] // rows=actual, cols=predicted; positive=Damaged

},

“perClass”: {

“Damaged”: {“precision”: float, “recall”: float, “f1”: float},

“NotDamaged”: {“precision”: float, “recall”: float, “f1”: float}

},

“macroF1”: float

}

PutBuildingPredictions(req) [source]

Persist per-building predictions from the interactive labeler.

The in-browser model predicts ``damaged`` (0/1) for every building. We

join those onto the layer’s cached building-footprints GeoPackage by row

index, write a predictions GeoPackage with the schema the reports expect

(``id`` row index, ``damaged``, ``damage_pct_0m``, ``unknown_pct``,

``area``), upload it, and set the embedding Model’s ``gpkgUrl`` so the

existing Validation/Assessment reports work unchanged.

Body: { projectId, imageLayerId, modelId,

predictions: [ { id, damaged, unknown? }, … ] }

PutBuildingValidation(req) [source]

Save (replace) building validation labels for an image layer.

Request body: BuildingValidation JSON

{

“projectId”: “…”,

“imageLayerId”: “…”,

“labels”: {

“<overture-id>”: {“id”: “…”, “label”: “Damaged|NotDamaged|Unknown”, “updatedAt”: “…”}

}

PutInteractiveLabels(req) [source]

Save the interactive labeler’s labels for an embedding model.

Stored in the INTERACTIVE_VALIDATION store keyed by modelId — separate

from the Building Validation (VALIDATION) store keyed by imageLayerId.

Body: { projectId, imageLayerId, modelId, labels: { <overture-id>: {…} } }

PutModelCatalog(req) [source]

Add a model to the model catalog for reuse as a base model.

This endpoint allows users to checkpoint successful training results and add them

to the model catalog with custom metadata. The catalogued models can then be used

as base models for future training runs. External models can also be added without

requiring validation of their existence in the HASTE system.

Returns:: func.HttpResponse: JSON response containing: - success (bool): Operation success status - catalogModel (dict): Complete catalog model object as stored - message (str): Success or error message

PutProject(req) [source]

Create a new project or update an existing project configuration.

This endpoint handles both project creation and updates using a unified PUT operation.

When creating a new project, a unique project ID and creation timestamp are automatically

generated if not provided. For updates, the existing project is modified with the

provided data while preserving unspecified fields.

Project validation ensures:

- Required fields are present and properly formatted

- Geographic coordinates are valid

- Classification schema is properly structured

- Event dates are valid ISO timestamps

Returns:: func.HttpResponse: JSON response containing: - success (bool): Operation success status - projectId (str): Project identifier (generated or provided) - message (str): Success or error message - project (dict): Complete project object as stored

PutRunEmbeddingQueueMessage(req) [source]

Queue a building-embedding job for the building labeling workflow.

Creates a Model with ``modelType=”embedding”`` (the embedding +

interactive-labeling sub-row entity) and enqueues it. Unlike training,

embedding needs no labels — only the image layer’s cached imagery and

building footprints (resolved later by the postprocessor).

UploadFileByChunk(req) [source]

Upload large files in chunks for efficient processing of geospatial datasets.

This endpoint handles chunked file uploads for large geospatial imagery files,

supporting resumable uploads and parallel chunk processing. Files are uploaded

in multiple chunks and assembled server-side for efficient handling of large

datasets that would be impractical to upload as single files.

The chunked upload process supports:

- Resumable uploads if connection is interrupted

- Progress tracking and validation

- Automatic file assembly upon completion

- Error handling for individual chunks

Returns:: func.HttpResponse: JSON response containing: - status (str): Upload status (‘success’, ‘partial’, ‘error’) - chunk_info (dict): Information about processed chunk - upload_progress (float): Overall upload progress percentage - file_info (dict): File metadata if upload is complete

HASTE Queue Functions

Contents

HASTE Queue Functions#

Architecture#

Azure Batch Pools#

Queue Triggers#

GetProcessImageLayerQueueMessage#

GetCreateModelRunQueueMessage#

GetRunInferenceQueueMessage#

UpdateStatsMessage#

GetArtifactsZipQueueMessage#

ImagePoisonQueueHandler#

Error Handling#

Configuration#

Azure Batch#

Container Registry#

Queue & Storage#

Deployment#

Development Setup#

Prerequisites#

Running Locally#

Local Debugging#

Auto-generated API Docs#