Utilities

The hastegeo.core.utils package provides shared utility modules used across the HASTE platform.

Logging (hastegeo.core.utils.logs)

Static logger utility for consistent logging across all HASTE components.

• Logger — Configurable logger with file and console handlers

class hastegeo.core.utils.logs.Logger

Bases: object

static add_console_handler(logger, level=logging.ERROR)

Adds a console handler to the logger.

Parameters:

• logger (Logger) – Logger instance.

• level (int) – Logging level.

static add_file_handler(logger, log_file, level=logging.INFO)

Adds a file handler to the logger.

Parameters:

• logger (Logger) – Logger instance.

• log_file (str) – File to log messages to.

• level (int) – Logging level.

static get_logger(name, log_file='app.log', log_dir=None, level=logging.INFO)

Creates and returns a logger instance.

Parameters:

• name (str) – Name of the logger.

• log_file (str) – File to log messages to.

• log_dir (str) – Directory to store log files.

• level (int) – Logging level.

formatter = logging.Formatter(

‘%(asctime)s - %(name)s - %(levelname)s - %(message)s - %(funcName)s - %(lineno)d’

)

static log_error(logger, message)

Logs an error message.

Parameters:

• logger (Logger) – Logger instance.

• message (str) – Error message to log.

static log_info(logger, message)

Logs an info message.

Parameters:

• logger (Logger) – Logger instance.

• message (str) – Info message to log.

static set_log_level(logger, level)

Sets the logging level for the given logger.

Parameters:

• logger (Logger) – Logger instance.

• level (int) – Logging level.

Metadata (hastegeo.core.utils.metadata)

Utility functions for ID generation, timestamps, and hashing.

• MetadataUtils — Static methods: generate_id(), generate_int_id(), get_timestamp(), get_short_date(), hash_string(), append_status_message()

class hastegeo.core.utils.metadata.MetadataUtils

Bases: object

static append_status_message(status_message, message, timestamp=None)

static generate_id()

static generate_int_id()

static generate_short_int_id(digits=4)

static get_short_date()

static get_timestamp()

static hash_string(string)

Queue Handler (hastegeo.core.utils.queues)

Azure Queue Storage client for sending and receiving async task messages.

• AzureQueueHandler — Queue operations: put_message(), get_messages(), delete_message()

class hastegeo.core.utils.queues.AzureQueueHandler(connection_string, queue_name, account_url)

Bases: object

__init__(connection_string, queue_name, account_url)

delete_message(message_id, pop_receipt=None)

get_message_by_id(message_id, pop_receipt=None)

get_messages(max_messages=1)

put_message(message, visibility_timeout=30)

Imagery (hastegeo.core.utils.imagery)

Geospatial imagery processing utilities using GDAL, rasterio, and OpenCV.

• ImageryUtils — Static methods for mosaic creation, COG generation, reprojection, and tile creation

Downloader (hastegeo.core.utils.downloader)

Multi-source imagery download client.

• ImageryDownloader — Downloads from HTTP URLs, Azure Blob Storage, and AWS S3 with retry logic

class hastegeo.core.utils.downloader.ImageryDownloader(sourceType, log_dir=None, log_file=None, log_level=INFO)

Bases: object

__init__(sourceType, log_dir=None, log_file=None, log_level=INFO)

download_aws_s3_files(urls, dst_directory)

download_azure_blob_files(urls, dst_directory)

download_imagery(**kwargs)

download_imagery_from_urls(urls, dst_directory)

download_maxar_imagery_post_pre(event_name, save_directory)

download_planet_imagery_post_pre(api_key, item_type, asset_type, pre_item_id, post_item_id, save_directory)

Data Utilities (hastegeo.core.utils.data)

Data transformation and file management helpers.

• extract_from_url() — Extract substring from URL using regex

• convert_json_to_geojson() — Transform JSON to GeoJSON FeatureCollection

• get_distinct_label_classes() — Extract unique classification labels

• directory_cleanup() — Delete directory and contents

• filter_roles() — Filter out anonymous/authenticated roles

hastegeo.core.utils.data.convert_json_to_geojson(input_json)

hastegeo.core.utils.data.directory_cleanup(directory)

hastegeo.core.utils.data.extract_from_url(url, pattern)

hastegeo.core.utils.data.filter_roles(roles)

hastegeo.core.utils.data.get_distinct_label_classes(label_data, key='primaryClass')

TensorBoard Parser (hastegeo.core.utils.tbparser)

Parse TensorBoard event logs to extract training metrics.

• parse_tb_event_logs() — Parse event logs into epochs, accuracy, loss

• calculate_metrics() — Calculate training progress metrics

hastegeo.core.utils.tbparser.calculate_metrics(logs, maxEpochs, time_field='elapsedDurationInMinutes', epoch_field='epoch')

Calculate completion metrics from TensorBoard logs.

Parameters:

• logs (str) – JSON string of logs containing epoch and time information.

• maxEpochs (int) – Maximum number of epochs.

• time_field (str) – The field name in logs that contains elapsed time information. Default is ‘elapsedDurationInMinutes’.

• epoch_field (str) – The field name in logs that contains epoch information. Default is ‘epoch’.

Returns:

A dictionary containing the following keys:

• ’completed_epochs’ (int): The number of completed epochs.

• ’approx_time_to_complete’ (float): The approximate time to complete the remaining epochs.

• ’total_elapsed_time’ (float): The total elapsed time.

• ’time_per_epoch’ (float): The average time per epoch.

Return type:

dict

hastegeo.core.utils.tbparser.parse_tb_event_logs(log_file_path)

Parses TensorBoard event logs to extract epoch, accuracy, and loss information.

Parameters:

log_file_path (str) – The file path to the TensorBoard event log file.

Returns:

A tuple containing:

• start_timestamp (str): The start timestamp of the events in UTC formatted as ‘%Y-%m-%d %H:%M:%S’.

• result (str): A JSON string containing a list of dictionaries, each representing an epoch with the following keys:

  • ’epoch’ (int): The epoch number.

  • ’elapsedDurationInMinutes’ (float): The duration of the epoch in minutes.

  • ’multiclassAccuracy’ (float): The multiclass accuracy for the epoch.

  • ’loss’ (float): The loss for the epoch.

  • ’wallTime’ (str): The wall time of the epoch event in UTC formatted as ‘%Y-%m-%d %H:%M:%S’.

Return type:

tuple

User Management (hastegeo.core.utils.user)

User invitation and email notification utilities.

• InvitationManager — Manage invitations via Azure Static Web Apps API and send emails via Azure Communication Services

• EmailSendResponse / CreateInvitationResponse — Response types

class hastegeo.core.utils.user.CreateInvitationResponse(email_id, invitation_link, error)

Bases: NamedTuple

email_id: str

Alias for field number 0

error: str | None

Alias for field number 2

invitation_link: str

Alias for field number 1

class hastegeo.core.utils.user.EmailSendResponse(email_id, sent, error)

Bases: NamedTuple

email_id: str

Alias for field number 0

error: str | None

Alias for field number 2

sent: bool

Alias for field number 1

class hastegeo.core.utils.user.InvitationManager(emails, roles=None, invite_config=None, invite_expire_hours=168, delete_existing=False)

Bases: object

__init__(emails, roles=None, invite_config=None, invite_expire_hours=168, delete_existing=False)

Initialize the InvitationManager with email addresses, user roles, and invitation expiration.

Parameters:

• emails (list[str]) – List of email addresses to invite.

• roles (list[str]) – List of roles to assign to invited users.

• invite_expire_hours (int, optional) – Number of hours before the invitation expires. Defaults to 1.

Raises:

EnvironmentError – If required environment variables are missing.

create_invitation(email)

Create an invitation for a single email address using Azure Static Web Apps REST API.

Parameters:

email (str) – The email address to invite.

Returns:

Dictionary with ‘success’ and ‘error’ keys containing invitation result or error details.

Return type:

dict

send_email(recipient_email, link)

send_invitations()

Validates emails, creates invitations, sends invitation emails, and returns the results.

For each email:

• If invalid, adds an Invite with error.

• If valid, attempts to create an invitation link via Azure Static Web Apps API.

  • If invitation creation fails, adds an Invite with error.

  • If invitation creation succeeds, sends an email with the link and records the send status.

Returns:

An Invites object containing Invite objects for all processed emails, with status and errors.

Return type:

Invites

validate_emails()

Validate the list of email addresses using email_validator.

Returns:

A tuple containing a list of valid emails and a list of invalid emails.

Return type:

tuple[list[str], list[str]]

class hastegeo.core.utils.user.UserManager(invite_config=None)

Bases: object

Manages user operations for Azure Static Web Apps including listing, deleting, and updating users.

__init__(invite_config=None)

Initialize the UserManager with configuration.

Parameters:

invite_config (InviteConfig) – Configuration for Static Web App connection

delete_user(user_id, auth_provider='aad')

Delete a user from the Azure Static Web App.

Parameters:

• user_id (str) – The ID of the user to delete

• auth_provider (str) – The authentication provider (default: “aad”)

Returns:

True if deletion was successful

Return type:

bool

Raises:

Exception – If deletion fails

delete_user_by_email(email, auth_provider='aad')

Delete a user by their email address.

Parameters:

• email (str) – Email address of user to delete

• auth_provider (str) – The authentication provider (default: “aad”)

Returns:

True if deletion was successful

Return type:

bool

Raises:

Exception – If user not found or deletion fails

find_user_by_email(email)

Find a user by their email address.

Parameters:

email (str) – Email address to search for

Returns:

User object if found, None otherwise

Return type:

dict | None

list_users()

List all users in the Azure Static Web App.

Returns:

List of users from the Static Web App

Return type:

list

Exceptions (hastegeo.core.utils.exceptions)

Custom exception types.

• UnsupportedFormatError — Raised for unsupported file formats

exception hastegeo.core.utils.exceptions.UnsupportedFormatError

Bases: Exception