Processors

The hastegeo.core.processors package contains the business logic layer for all HASTE operations. Processors coordinate between data layers, runners, and utilities to implement the platform’s core functionality.

Imagery (hastegeo.core.processors.imagery)

Handles satellite imagery download, preprocessing, and Cloud Optimized GeoTIFF creation.

• ImageryPreProcessor — Queue imagery for async processing

• ImageryPostProcessor — Execute the full imagery preprocessing pipeline (download, mosaic, COG creation, tiling)

• ImageryLogRecord — Timestamped log record for imagery processing

class hastegeo.core.processors.imagery.ImageryLogRecord(timestamp, message)

Bases: NamedTuple

Named tuple representing a log record for imagery processing operations.

This structure captures timestamped log messages generated during imagery preprocessing workflows, providing structured logging for debugging and progress tracking.

timestamp

ISO formatted timestamp when the log message was created

Type:

str

message

Descriptive log message about the processing operation

Type:

str

Example

```python log_record = ImageryLogRecord(

timestamp=”2023-08-01T12:00:00Z”, message=”Starting imagery preprocessing for layer img_123”

) print(log_record) # “2023-08-01T12:00:00Z: Starting imagery preprocessing…” ```

message: str

Alias for field number 1

timestamp: str

Alias for field number 0

class hastegeo.core.processors.imagery.ImageryPostProcessor(image_data, config=None)

Bases: object

__init__(image_data, config=None)

process()

class hastegeo.core.processors.imagery.ImageryPreProcessor(image_data, config=None)

Bases: object

Processor for handling satellite and aerial imagery preprocessing workflows.

This class manages the complete imagery preprocessing pipeline including download, validation, mosaic creation, and Cloud Optimized GeoTIFF (COG) generation. It integrates with Azure Batch services for scalable processing and Azure Queue Storage for asynchronous job management.

The processor handles both pre-event and post-event imagery for change detection and damage assessment workflows, supporting multiple imagery sources and formats.

Parameters:

• image_data (ImageLayer) – Image layer model containing imagery metadata and URLs

• config (Config, optional) – Configuration object with environment settings. Defaults to None, which creates a new Config instance.

image_data

The image layer being processed

Type:

ImageLayer

logger

Logger instance for tracking processing operations

Type:

Logger

queue

Queue handler for asynchronous job management

Type:

AzureQueueHandler

config

Configuration object with system settings

Type:

Config

Raises:

ValueError – If image_data is not a valid ImageLayer instance

Example

```python # Create image layer with pre and post event imagery image_layer = ImageLayer(

imageLayerId=”layer_123”, projectId=”proj_456”, preEventImageryUrls=[”https://storage/pre_event.tif”], postEventImageryUrls=[”https://storage/post_event.tif”]

)

# Initialize processor and queue for processing processor = ImageryPreProcessor(image_layer) processor.queue_for_processing() ```

Note

The processor supports various imagery sources including Sentinel-2, Landsat, WorldView, and other satellite/aerial imagery formats. All processing operations are designed to handle large-scale imagery datasets efficiently through cloud-based batch processing.

__init__(image_data, config=None)

queue_for_processing()

Queue the imagery layer for asynchronous preprocessing.

This method prepares the image layer for processing by setting initial status values and sending the layer data to the Azure Queue for asynchronous processing by background workers.

The method sets up progress tracking with 4 total processing steps: 1. Download and validation of imagery files 2. Mosaic creation for multi-file imagery 3. Cloud Optimized GeoTIFF (COG) generation 4. Metadata updates and finalization

Updates the image layer status to PENDING and resets progress indicators before queuing the processing job.

Returns:

The updated image layer with processing status set to PENDING

Return type:

ImageLayer

Example

`python processor = ImageryPreProcessor(image_layer) updated_layer = processor.queue_for_processing() print(f"Layer {updated_layer.imageLayerId} queued with status: {updated_layer.status}") `

Note

The actual processing is handled asynchronously by Azure Queue consumers. This method only queues the job and updates initial status.

hastegeo.core.processors.imagery.save_imagery_layer(image_layer, config=None, prepare_footprints=False)

Persist imagery/label work without overwriting footprint job state.

Only the final successful save may publish footprint work. Nothing in the imagery handler saves a stale footprint snapshot after publication. Other imagery saves (including error handling/redelivery) also exclude footprint-owned fields, using MetadataProcessor’s top-level merge.

Training (hastegeo.core.processors.train)

ML model training workflow management.

• BaseTrainProcessor — Base class for training operations

• TrainPreprocessor — Queue models for training (prepares experiment config)

• TrainPostprocessor — Execute and monitor training jobs on Azure Batch

class hastegeo.core.processors.train.BaseTrainProcessor(model, image_layer, config=None)

Bases: object

__init__(model, image_layer, config=None)

class hastegeo.core.processors.train.TrainPostprocessor(model, image_layer=None, label_project=None, project=None, config=None)

Bases: BaseTrainProcessor

__init__(model, image_layer=None, label_project=None, project=None, config=None)

cancel()

process()

class hastegeo.core.processors.train.TrainPreprocessor(model, config=None)

Bases: object

__init__(model, config=None)

send_to_queue(status=None)

Inference (hastegeo.core.processors.inference)

Model inference execution and monitoring.

• BaseInferenceProcessor — Base class for inference operations

• InferencePreprocessor — Prepare and queue model inference requests

• InferencePostprocessor — Execute and monitor inference jobs

• InferenceLogRecord — Timestamped inference log entry

class hastegeo.core.processors.inference.BaseInferenceProcessor(model, image_layer, config=None)

Bases: object

__init__(model, image_layer, config=None)

class hastegeo.core.processors.inference.InferenceLogRecord(timestamp, message)

Bases: NamedTuple

message: str

Alias for field number 1

timestamp: str

Alias for field number 0

class hastegeo.core.processors.inference.InferencePostprocessor(model, image_layer=None, experiment_config=None, config=None)

Bases: BaseInferenceProcessor

__init__(model, image_layer=None, experiment_config=None, config=None)

cancel()

process()

class hastegeo.core.processors.inference.InferencePreprocessor(model, config=None)

Bases: object

__init__(model, config=None)

send_to_queue()

Labels (hastegeo.core.processors.labels)

Labeling task generation from imagery.

• LabelTaskGenerator — Generate labeling tasks from imagery layers with optional grid-based subdivision

Metadata (hastegeo.core.processors.metadata)

High-level metadata operations over the data layer.

• MetadataProcessor — CRUD operations for project/model/user metadata with support for multiple storage backends

class hastegeo.core.processors.metadata.MetadataProcessor(data_type, partition_key=None, config=None)

Bases: object

Processor for managing metadata storage and retrieval operations.

The MetadataProcessor provides a high-level interface for working with metadata in various storage backends. It handles data serialization, storage operations, and provides convenient methods for common metadata operations like saving, loading, and listing records.

Parameters:

• data_type (str) – Type of metadata being processed (e.g., ‘project’, ‘model’, ‘labels’).

• partition_key (str, optional) – Partition key for data organization. Defaults to None.

• config (Config, optional) – Configuration instance. If None, creates a new Config instance. Defaults to None.

storage

Underlying storage layer instance.

Type:

UnifiedDataLayer

data_type

The metadata type this processor handles.

Type:

str

Example

>>> processor = MetadataProcessor('project', 'user_123')
>>> processor.save('project_1', {'name': 'My Project'})
>>> project = processor.load('project_1')

__init__(data_type, partition_key=None, config=None)

Initialize the metadata processor.

Parameters:

• data_type (str) – Type of metadata being processed.

• partition_key (str, optional) – Partition key for data organization. Defaults to None.

• config (Config, optional) – Configuration instance. Defaults to None.

build_url(key, data_format='json')

Build a remote URL for a record without a per-item existence check.

Intended for callers that have already confirmed the record exists in bulk via list_keys(). On the blob backend this avoids a network round-trip per key (the container read SAS is shared), so it is a local operation and is not counted as a storage round-trip.

delete(key, data_format='json')

Delete metadata from the backend storage.

delete_all_from_partition()

Delete all metadata from the backend storage.

export(key, data_format='json')

Export metadata to a file in the specified format.

Parameters:

• key (str) – Unique identifier for the metadata record.

• data_format (str, optional) – Format for data serialization. Defaults to “json”.

Returns:

File path where the metadata is exported.

Return type:

str

list_keys(data_format='json')

List the identifiers present for this data type in the partition.

A cheap, metadata-only alternative to load_all_from_partition for bulk existence checks (does not download record contents).

load(key, data_format='json')

Load metadata from the backend storage.

load_all(data_format='json')

Load all metadata from the backend storage.

load_all_from_partition(data_format='json')

Load all metadata from the backend storage.

load_and_combine_sub_data_types(key, data_types)

Load and combine metadata from multiple data types.

load_bounded(max_records, data_format='json')

load_filtered(predicate, data_format='json')

Load partition records matching every key/value in predicate.

This is an explicit client-side fallback: the partition is loaded once and then filtered in process. Backends need a separate query primitive before this can reduce transferred records.

load_map(keys, data_format='json', max_workers=None)

Load many records by key concurrently.

Returns {key: record}; a key whose record is missing maps to None (mirrors a per-key load that raises FileNotFoundError). Bounded parallelism overlaps per-key storage operations while the process-wide executor caps aggregate concurrency. Preserves perf instrumentation across worker threads by binding the active counter inside each task.

load_page(page, page_size, data_format='json', target=None, status=None, project_id=None, max_records=None)

save(key, metadata, data_format='json')

Save or upsert metadata to the backend storage.

This method performs an upsert operation - if metadata with the given key already exists, it will be updated; otherwise, new metadata will be created.

Parameters:

• key (str) – Unique identifier for the metadata record.

• metadata (dict) – Metadata object to save.

• data_format (str, optional) – Format for data serialization. Defaults to “json”.

Example

>>> processor.save('project_1', {'name': 'My Project', 'status': 'active'})

Statistics (hastegeo.core.processors.stats)

Project statistics aggregation.

• StatsPreProcessor — Queue stats update requests

• StatsPostProcessor — Process stats updates (add/update/delete project entries)

class hastegeo.core.processors.stats.StatsPostProcessor(request, summary=None, config=None)

Bases: object

__init__(request, summary=None, config=None)

update()

class hastegeo.core.processors.stats.StatsPreProcessor(request, summary=None, config=None)

Bases: object

__init__(request, summary=None, config=None)

send_to_queue()

Artifacts (hastegeo.core.processors.artifacts)

Model artifact management, packaging, and download.

• ArtifactProcessor — Manages model artifacts including zipping, downloading, and Azure Batch zip job submission

class hastegeo.core.processors.artifacts.ArtifactProcessor(partition_key=None, config=None, model=None, model_artifacts=None)

Bases: object

__init__(partition_key=None, config=None, model=None, model_artifacts=None)

fetch_artifact(identifier=None, extra_partition_keys=None, src_path=None, dst_path=None)

Download the artifact from the storage.

get_download_url(identifier=None, artifact_path=None, extra_partition_keys=None)

Get the download URL for the artifact.

prepare_zip_job()

process_zip()

send_to_zip_queue()

Put a message to the queue.

store_artifact(artifact_name, data=None, src_path=None, namespace=None)

Store an artifact in artifact storage.

submit_zip_job()

Uploader (hastegeo.core.processors.uploader)

Chunked file upload handling.

• FileUploader — Process chunked file uploads with chunk storage and final assembly

class hastegeo.core.processors.uploader.FileUploader(project_id, config=None)

Bases: object

__init__(project_id, config=None)

delete_chunk(chunk_key)

finalize(file_id, total_chunks, data_format=None)

get_chunk(chunk_key)

save_chunk(chunk_number, chunk_data, file_id, total_chunks, action='add', data_format=None)