AIME API Server Documentation

class api_server.api_server.APIServer(api_name)

AIME API Server

Parameters:: api_name (str) – Name of API server

load_server_configuration(app): Parses server configuration file.

async worker_check_server_status(request)

Route /worker_check_server_status for API worker interface to check server connection and queue validation for given job type and worker authentication.

Parameters:: request (sanic.request.types.Request) – API worker interface request containing worker validation
Returns:: _description_
Return type:: sanic.response.types.JSONResponse

Examples

Example successful response json to API worker interface:

response.json() = {'cmd': 'ok'}

Example response json if AIME ML API replied with warning:

response.json() = {
    'cmd': 'error',
    'msg': 'Message'
}

Example response json if AIME ML API replied with warning:

response.json() = {
    'cmd': 'warning',
    'msg': 'Message'
}

async worker_job_progress(request)

Receive progress results from api worker interface via route /worker_job_progress and put it to APIServer.progress_states[job_id] to make it available for APIEndpoint.api_progress().

Parameters:: request (sanic.request.types.Request) – API worker interface request containing progress results
Returns:: Response to API worker interface with ‘cmd’: ‘ok’ when API server received data.
Return type:: sanic.response.types.JSONResponse

Examples

Examples request payload from API worker interface:

request.json = [{
    'job_id': 'JID1',
    'progress': 50,
    'progress_data': {'progress_images': ['base64-string', 'base64-string', ...]},
    'start_time_compute': 1700424731.9582381,
    'start_time': 1700424731.952994
}]

async worker_job_request_json(request)

Request from API worker interface for a job on route /worker_job_request. If the worker is validated and a job is put in the job_queue within job_timeout the job data including ‘cmd’: ‘job’ is received from the job queue and send to API worker interface in the json response. If there is no job put in the within job_timeout ‘cmd’: ‘no_job’ is received from the job queue and send to API worker interface in the json response. If the validation failed ‘cmd’: ‘error’, or ‘warning’ with ‘msg’ is sent to API worker interface.

Parameters:: request (sanic.request.types.Request) – Job request from API worker interface on route /worker_job_request.
Returns:: Response to API worker interface with job data or messages.
Return type:: sanic.response.types.JSONResponse

Examples

Example response json if client requested job:

response.json() = {
    'cmd': 'job',
    'endpoint_name': 'stable_diffusion_xl_txt2img',
    'wait_for_result': False,
    'job_data': {
        'job_id': 'JID1',
        'start_time': 1700424731.952994,
        'start_time_compute': 1700424731.9582381
        'prompt': 'prompt',
    },
    'progress_descriptions': = {
        'progress_images': {
            'type': 'image_list',
            'format': 'JPEG',
            'color_space': 'RGB'
        }
    }
    'output_descriptions': = {
        'images': {
            'type': 'image_list',
            'format': 'JPEG',
            'color_space': 'RGB'
        },
        'seed': {'type': 'integer'},
        'prompt': {'type': 'string'},
        'error': {'type': 'string'}
    }

}

Example response json if no client requested job within job_timeout:

response.json() = {
    'cmd': 'no_job',
}

Example response json if AIME ML API replied with error:

response.json() = {
    'cmd': 'error',
    'msg': 'Message'
}

Example response json if AIME ML API replied with warning:

response.json() = {
    'cmd': 'warning',
    'msg': 'Message'
}

async worker_job_result_json(request)

Receive final job result from API worker interface via route /worker_job_result. Processes the result and puts it to the future related to the job initialized in APIEndpoint.api_request to make it available by APIEndpoint.api_request and APIEndpoint.api_progress to send it to the client.

Parameters:: request (sanic.request.types.Request) – API worker interface request containing the job results.
Returns:: Response to API worker interface with ‘cmd’: ‘ok’ when API server received data.
Return type:: sanic.response.types.JSONResponse

class api_server.api_endpoint.APIEndpoint(app, config_file)

AIME API endpoint

Parameters:

app (APIServer) – Instance of APIServer()
config_file (str) – Endpoint config file path.

async api_progress(request)

Client request on route /self.endpoint_name/progress called periodically by the client interface to receive progress and end results if the input parameter ‘wait_for_result’ of the related api_request was False. Takes the progress results from APIServer.progress_states put there by APIServer.worker_job_progress() and sends it as response json to the client. When the job is finished the final job result is awaited and taken from the job queue in finalize_request() and sent as response json to the client.

Parameters:: request (sanic.request.types.Request) – Request with client_session_auth_key from client to receive progress and end result
Returns:: Response to client with progress result and end result
Return type:: sanic.response.types.JSONResponse

async api_request(request)

Client request on route /self.endpoint_name with input parameters for the workers related to the job type given in the input parameters. The client input parameters are validated and prepared for the worker job data. Next an asyncio.Future() for the job is initialized where the result will be put by APIServer.worker_job_result_json() when the job is finished. The job data is put to the job type related asyncio.Queue() to be accessed by the workers in APIServer().worker_job_request(). If ‘wait_for_result’ is True the end result of the job is awaited via finalize_request and returned to the client. If False, the client gets a quick confirmation response and the result can be requested on route /self.endpoint_name/progress in the parameter ‘job_result’.

Parameters:: request (sanic.request.types.Request) – Request from client
Returns:: Response to client
Return type:: sanic.response.types.JSONResponse

async client_login(request)

Route for client interface to login to the API Server while receiving a client session authentication key. :param request: Request from client. :type request: sanic.request.types.Request

Returns:: Response to client containing the client session authentication key.
Return type:: sanic.response.types.JSONResponse

async finalize_request(request, job_id, job_future)

Awaits the result until the APIServer.worker_job_result_json() puts the job results in the related future initialized in api_request() to get the results.

Parameters:

request (sanic.request.types.Request) – _description_
job_id (_type_) – _description_
job_future (_type_) – _description_

Returns:

Dictionary representation of the response.

Return type:

dict

get_endpoint_description()

Loads endpoint description parameters title, name, description, client_request_limit, http_methods from given endpoint config file.

Parameters:: config (dict) – Config dictionary
Returns:: Tuple with endpoint descriptions title, name, description, client_request_limit, http_methods.
Return type:: tuple (str, str, str, int, list)

get_http_methods(ep_config)

Loads http methods defined in given endpoint config file

Parameters:: config (dict) – Config dictionary
Returns:: List of http methods like [‘GET’, ‘POST’]
Return type:: list

get_param_config()

Parses endpoint input-, output-, progress-, and session-parameter configuration from given endpoint config file.

Parameters:

config (dict) – Config dictionary

Returns:

Tuple of dictionaries with input-, output-, progress-,: and session-parameter configuration

Return type:

tuple (dict, dict, dict, dict)

get_worker_params()

Parses worker parameters like job type and worker authorization key from endpoint config file.

Parameters:: config (dict) – Config dictionary
Returns:: Tuple of job_type and worker_auth_key
Return type:: tuple (str, str)

async validate_input_parameters_for_job_data(input_args): Check if worker input parameters received from client are as specified in the endpoint config file

class api_server.job_queue.JobQueue(job_type, worker_auth_key)

Job queue to manage jobs for the given job type. Assignes jobs offered from client on APIEndpoint.api_request via route /endpoint_name with a job_id and collects the job_data. The workers asking for jobs on worker_job_request_json get the job_data for the next job in the queue to process. Also monitors states of workers in registered_workers and mean_job_durations.

Parameters:

job_type (str) – Job type
worker_auth_key (str) – Key for worker authorization

fetch_waiting_job()

Get job data of the next waiting job in the queue. Returns None if no job is waiting

Returns:: Job data of the next job in the queue, None if no job available
Return type:: dict

async get(job_timeout=60)

Get job data of the next job in the queue. Timeout after self.job_timeout

Returns:: Job data of the next job in the queue
Return type:: dict

async put(job_data)

Put job_data from the client job offer to the job_queue.

Parameters:: job_data (dict) – Job data of the job
Returns:: _description_
Return type:: _type_

class api_server.utils.ffmpeg.FFmpeg(param_name, base64_string=None, config={}, errors=None, media_binary=None, arg_type=None, resize_method=None, input_temp_file_config=None, output_temp_file_config=None, check_conversion_config=False)

Python binding for FFmpeg to analyze and convert media data.

Parameters:

param_name (str) – Name of the input parameter.
base64_string (str, optional) – String of base64 encoded input parameter. Defaults to None.
config (dict, optional) – Configuration of parameter containing arg_type, resize_method, input_temp_file_config, output_temp_file_config and check_conversion_config. Defaults to {}.
errors (list, optional) – Error list to accumulate errors occuring during media analysis or conversion. If not given, errors will be raise. Defaults to None.
media_binary (bytes, optional) – Binary data of input data as alternative to base64_string. Defaults to None.
arg_type (str, optional) – Input parameter type. Must be present if no config is given. Supported values: (‘image’, ‘audio’, ‘video’). Defaults to None.
resize_method (str, optional) – Resize method for input image conversion. Supported values: (‘scale’, ‘crop’). Defaults to None.
input_temp_file_config (str, optional) – Config whether an input temp file is generated. Supported values: (‘auto’, ‘yes’, ‘no’). Defaults to None.
output_temp_file_config (str, optional) – Config whether an output temp file is generated. Supported values: (‘auto’, ‘yes’, ‘no’). Defaults to None.
check_conversion_config (bool, optional) – Config whether to check if the media attributes of the input parameter are as supposed after its conversion. Defaults to False.

async analyze(base64_string=None, media_binary=None)

Analyze input parameter with ffprobe. Save the measured media parameters in self.media_params and return it.

Parameters:

base64_string (str, optional) – String of base64 encoded input parameter. Defaults to None.
media_binary (bytes, optional) – Binary data of input data as alternative to base64_string. Defaults to None. Defaults to None.

Returns:

Measured media parameters as instance of MediaParams.

Return type:

MediaParams

async convert(target_media_params, resize_method=None)

Convert input parameter with ffmpeg to given target_media_paramters

Parameters:

target_media_params (MediaParams or dict) – Target media parameters for conversion.
resize_method (str, optional) – Resize method for input image conversion to change resize method given at initialization. Supported values: (‘scale’, ‘crop’). Defaults to None.

async get_data(output_format='base64')

Get the base64 or binary representation of the stored media data

Parameters:: output_format (str, optional) – Desired output format. Supported values: (‘base64’, ‘binary’). Defaults to ‘base64’.
Returns:: Base64 or binary representation of stored media data.
Return type:: str or bytes

class api_server.utils.ffmpeg.MediaParams: Specialized dictionary class designed to store media parameter information. It inherits from Python’s built-in dict class and predefines keys related to the media attributes.

class api_server.input_validation.InputValidationHandler(input_params, ep_input_param_config, server_input_param_config)

Handler to validate and convert API server input parameters.

Parameters:

input_params (dict) – Dictionary containing all input parameters.
ep_input_param_config (dict) – Endpoint configuration of all input parameters.
server_input_param_config (dict) – Server configuration of all input parameters.

async validate_input_parameters()

Validate all input parameters.

Returns:: Tuple of dictionary containing valid parameters and list of validation errors.
Return type:: tuple(dict, list)

async validate_media_base64_string(media_base64)

Validation of given base64 representation of media input parameter. Checks the media attributes with ffprobe and validates it with the server and the endpoint configuration. If auto_convert is True in endpoint configuration, media input parameter will be converted to valid target media attributes with ffmpeg defined in the endpoint config.

Parameters:: media_base64 (str) – Base64 representation of media input parameter
Returns:: Base64 representation of validated and converted media input parameter
Return type:: str