Tutoriales

Ejemplos de uso Python

Dependencias comunes

import io
import os
import requests

Definir paths de la API para referencia

Definimos las rutas de la API REST que vamos a utilizar:

class Paths:
    def __init__(self):
        self.files_path = f"{HOST}/v1/files"
        self.runs_path = f"{HOST}/v1/runs"

Subida de ficheros

Definición de funciones para la subida de los ficheros. Estos métodos devuelven un file_id (o listado de file_id) identificando los ficheros en la plataforma. Estos se utilizarán en otras llamadas.

def upload_file(app_paths: Paths, api_key: str, src_path: str, dest_path: str) -> str:
    logger.info(f"Uploading file {src_path}")
    with open(src_path, "rb") as f:
        image_bytes = f.read()
    image_stream = io.BytesIO(image_bytes)

    response = requests.post(
        url=app_paths.files_path,
        headers={"X-API-Key": f"{api_key}"},
        files={"files": (dest_path, image_stream, "multipart/form-data")},
    )
    assert response.status_code == 201, response.text
    return response.json().get("file_ids")[0]

def upload_directory(app_paths: Paths, api_key: str, src_path: str, dest_path: str) -> Dict[str, str]:
    if not os.path.isdir(src_path):
        logger.error("Source path should be a directory, not a file. Returning...")
        return []
    result: Dict[str, str] = {}
    for root, dirs, files in os.walk(src_path):
        for file in files:
            file_src_path = os.path.join(root, file)
            file_dest_path = os.path.join(dest_path, file)
            result[file] = upload_file(app_paths, api_key, file_src_path, file_dest_path)
            logger.info(f"File {file} uploaded to Veraquo")
    return result

Procesamiento de los ficheros

Definición de funciones para procesar los ficheros. Estos métodos devuelven un run_id (o listado de run_id) identificando cada ejecución en la plataforma. Estos se utilizarán en otras llamadas para consultar su estado, obtener los resultados, etc.

def process_file(paths: Paths, api_key: str, file_id: str):
    # check file status
    file_info = requests.get(
        url=f"{paths.files_path}/{file_id}",
        headers={"X-API-Key": f"{api_key}"},
    )
    
    file_format: str = file_info.json().get("format")
    status = file_info.json().get("state")
    if status != "ready":
        print(f"Error: file {file_id} not ready")
    
    payload = {"file_ids": [file_id]}

    processing_response = requests.post(
            url=f"{paths.runs_path}/detection/{file_format}",
            headers={"X-API-Key": f"{api_key}"},
            json=payload,
        )
    run_id = inference_response.json().get("run_ids")[0]
    return run_id

Definimos una función para comprobar el estado del procesado:

def get_processing_status(paths: Paths, api_key: str, run_id: str):
    status_response = requests.get(
        url=f"{paths.runs_path}/{run_id}/status",
        headers={"X-API-Key": f"{api_key}"},
    )
    response_json = status_response.json()
    run_status = response_json.get("status")
    return run_status

Definimos una función para obtener los resultados de un procesado. Podemos obtener la información del analisis de la información del fichero:

def get_processing_results(paths: Paths, api_key: str, file_id: str):
    response = requests.get(url=f"{paths.files_path}/{file_id}", headers={"X-API-Key": f"{api_key}"})
    file_details = response.json()
    detection_results = file_details["detection_results"]
    return detection_results

Eliminación de los ficheros

Definimos una función para borrar los ficheros después de su procesado:

def delete_files_from_storage(paths: Paths, api_key: str, file_ids: List[str]):
    for file_id in file_ids:
        delete_url = f"{paths.files_path}/{file_id}"
        response = requests.delete(url=delete_url, headers={"X-API-Key": f"{api_key}"},)

        if response.status_code == 204:
            print(f"File {file_name} deleted from storage")
        else:
            print(f"Problem deleting file {file_name}: {response.text}")