# Python Integration Guide

This page provides a complete, ready-to-use Python script for integrating the LeapAI CXR API into your application. The script demonstrates health checks, single inference, and bulk inference.

# Prerequisites

Install the required Python package:

pip install requests

# Configuration

Update the following constants in the script with your environment details:

BASE_URL = "http://<server-ip>:8500"
TOKEN = "your_jwt_token"

# Complete Integration Script

"""
LeapAI CXR - Python Integration Script
======================================
A reference implementation for interacting with the
LeapAI CXR inference API.

Usage:
    python leapai_cxr_client.py

Requirements:
    pip install requests
"""

import os
import json
import requests
from typing import Optional


# =============================================================================
# Configuration
# =============================================================================

BASE_URL = "http://localhost:8500"  # LeapAI CXR server URL
TOKEN = "your_jwt_token"            # JWT token provided post-installation


# =============================================================================
# Health Check
# =============================================================================

def health_check(base_url: str) -> bool:
    """
    Check if the LeapAI CXR server is running and responsive.
    """
    try:
        response = requests.get(f"{base_url}/ping")
        if response.status_code == 200:
            data = response.json()
            print(f"[OK] Server is healthy: {data}")
            return True
        print(f"[ERR] Server returned status {response.status_code}")
        return False
    except requests.exceptions.ConnectionError:
        print(f"[ERR] Server is not reachable at {base_url}")
        return False


# =============================================================================
# Single Inference
# =============================================================================

def run_single_inference(
    base_url: str,
    token: str,
    input_image_path: str,
    save_to_disk: bool,
    patient_id: Optional[str] = None,
    output_folder_path: Optional[str] = None,
    timeout: int = 300,
) -> dict:
    """
    Run AI inference on a single DICOM image.

    If save_to_disk is True, output files are written under output_folder_path.
    If save_to_disk is False, source_image and heatmap are base64-encoded PNG strings
    in the response.
    """
    url = f"{base_url}/single_inference"
    payload = {
        "input_image_path": input_image_path,
        "save_to_disk": save_to_disk,
    }
    if patient_id:
        payload["patient_id"] = patient_id
    if save_to_disk:
        payload["output_folder_path"] = output_folder_path

    headers = {
        "Content-Type": "application/json",
        "Authorization": token,  # NOTE: No "Bearer" prefix
    }

     try:
        response = requests.post(url, json=payload, headers=headers, timeout=timeout)
        response_data = response.json()

        if response.status_code == 200 and response_data.get("Result") == "Success":
            print(f"[✓] Single inference completed successfully.")
        else:
            print(f"[✗] Single inference failed: {response_data.get('Message')}")

        return response_data

    except requests.exceptions.Timeout:
        print("[✗] Request timed out. The image may be too large.")
        return {"Result": "Failure", "Message": "Request timeout"}
    except Exception as e:
        print(f"[✗] Single inference error: {e}")
        return {"Result": "Failure", "Message": str(e)}


# =============================================================================
# Bulk Inference
# =============================================================================

def run_bulk_inference(
    base_url: str,
    token: str,
    input_folder_path: str,
    output_folder_path: str,
    timeout: int = 3600,
) -> dict:
    """
    Run AI inference on all DICOM images in a folder.
    """
    url = f"{base_url}/bulk_inference"
    payload = {
        "input_folder_path": input_folder_path,
        "output_folder_path": output_folder_path,
    }
    headers = {
        "Content-Type": "application/json",
        "Authorization": token,  # NOTE: No "Bearer" prefix
    }

    try:
        print(f"[…] Starting bulk inference on: {input_folder_path}")
        print(f"    Output will be saved to:    {output_folder_path}")
        print(f"    Timeout: {timeout}s. This may take a while...")

        response = requests.post(url, json=payload, headers=headers, timeout=timeout)
        response_data = response.json()

        if response.status_code == 200 and response_data.get("Result") == "Success":
            print(f"[✓] Bulk inference completed successfully.")
        else:
            print(f"[✗] Bulk inference failed: {response_data.get('Message')}")

        return response_data

    except requests.exceptions.Timeout:
        print("[✗] Request timed out. Consider increasing the timeout for large batches.")
        return {"Result": "Failure", "Message": "Request timeout"}
    except Exception as e:
        print(f"[✗] Bulk inference error: {e}")
        return {"Result": "Failure", "Message": str(e)}


# =============================================================================
# Read Prediction Results
# =============================================================================

def read_prediction_json(json_path: str) -> Optional[dict]:
    """
    Read and parse a prediction JSON file generated by the inference engine.
    """
    try:
        with open(json_path, "r") as f:
            data = json.load(f)

        print("\n" + "=" * 60)
        print(f"Prediction Results: {os.path.basename(json_path)}")
        print("=" * 60)
        print(f"Overall Result : {data.get('result', 'N/A')}")
        print(f"Patient ID     : {data.get('metadata', {}).get('patient_id', 'N/A')}")
        print(f"Patient Name   : {data.get('metadata', {}).get('patient_name', 'N/A')}")
        print("=" * 60)
        print(f"{'Condition':<25} {'Detected':<10} {'Confidence':<12}")
        print("-" * 47)

        for finding in data.get("findings", []):
            name = finding.get("name", "Unknown")
            presence = "YES" if finding.get("presence") else "NO"
            confidence = finding.get("confidence", 0.0)
            print(f"{name:<25} {presence:<10} {confidence:.4f}")

        print("=" * 60)
        print(f"Heatmap      : {data.get('heatmap', 'N/A')}")
        print(f"Source Image : {data.get('source_image', 'N/A')}")
        print("=" * 60 + "\n")

        return data

    except FileNotFoundError:
        print(f"[ERR] File not found: {json_path}")
        return None
    except json.JSONDecodeError:
        print(f"[ERR] Invalid JSON: {json_path}")
        return None


# =============================================================================
# Main - Example Usage
# =============================================================================

def main():
    """
    Example workflow demonstrating the full LeapAI CXR integration:
      1. Health check
      2. Run single inference
      3. Run bulk inference (optional)
      4. Parse prediction results
    """

    print("\n" + "=" * 60)
    print("LeapAI CXR - Python Integration Example")
    print("=" * 60 + "\n")

    # Step 1: Health Check
    print("[Step 1] Checking server health...")
    if not health_check(BASE_URL):
        print("Server is not available. Exiting.")
        return

    # Step 2: Single Inference
    print("\n[Step 2] Running single inference...")
    single_result = run_single_inference(
        base_url=BASE_URL,
        token=TOKEN,
        input_image_path="C:/data/patient001.dcm",
        save_to_disk=True,
        patient_id="PAT001",
        output_folder_path="C:/output/",
    )
    print(json.dumps(single_result, indent=2))

    # Step 3: Bulk Inference (Optional)
    print("\n[Step 3] Running bulk inference...")
    bulk_result = run_bulk_inference(
        base_url=BASE_URL,
        token=TOKEN,
        input_folder_path="C:/data/dicom/batch_001/",
        output_folder_path="C:/output/batch_001/",
        timeout=3600,
    )
    print(json.dumps(bulk_result, indent=2))

    # Step 4: Read Prediction Results
    print("\n[Step 4] Reading prediction results...")
    prediction_file = "C:/output/<patient_id>/<patient_id>_predictions.json"
    read_prediction_json(prediction_file)


if __name__ == "__main__":
    main()

# Function Reference

Function	Description
`health_check(...)`	Verify the server is running via `/ping`
`run_single_inference(...)`	Process a single DICOM image
`run_bulk_inference(...)`	Process a folder of DICOM images
`read_prediction_json(...)`	Parse and display a `_predictions.json` file

# Key Notes

Authorization Header

The Authorization header must contain the raw JWT token - do not prefix it with Bearer.

# Correct
headers = {"Authorization": "eyJhbGciOiJIUzI1NiIsIn..."}

# Incorrect
headers = {"Authorization": "Bearer eyJhbGciOiJIUzI1NiIsIn..."}

Timeout for Bulk Inference

Bulk inference can take a long time depending on the number of DICOM files. The default timeout is set to 3600 seconds (1 hour). Adjust the timeout parameter as needed for your dataset size.

save_to_disk = false

When save_to_disk is false, the API response includes source_image and heatmap as base64-encoded PNG strings. No files are written to disk.

All file paths must be absolute paths on the server where LeapAI CXR is running. If your client application is on a different machine, the paths refer to locations on the server's filesystem, not the client's filesystem.