earth-external-fitness 0.0.5

Remote fitness execution service for the Earth GA platform, processing individual calculation requests.

earth-external-fitness

A Python SDK for connecting custom local or remote compute resources to Genetic Algorithm (GA) experiments on the Earth Platform.

This SDK abstracts the infrastructure of authentication, heartbeat management, and population polling, allowing you to focus strictly on implementing your fitness calculation logic.

Key Features

• State-Aware Polling: Automatically adjusts polling frequency based on the experiment's lifecycle (Initializing -> Running -> Completed).
• Asynchronous Heartbeats: Maintains the active status of your worker to prevent experiment abortion by the platform.
• Queue-Based Distribution: Efficiently fetches individuals and distributes them for concurrent or parallel processing.
• Production Ready: Built-in support for JWT authentication (Keycloak) and thread-safe token refreshing.
• Graceful Termination: Automatically shuts down when the experiment reaches a terminal state (Completed, Aborted, Error, or Paused).

Installation

uv add earth-external-fitness
# OR
pip install earth-external-fitness

Basic Usage

The most common way to use the SDK is via an async loop.

import asyncio
from earth_external_fitness import EarthExternalFitnessSDK, EarthSDKSettings, FitnessCalculationResult

async def main():
    # 1. Configure settings directly
    settings = EarthSDKSettings(
        gateway_url="https://earth.evolution.inc/api/external-fitness",
        username="your_username",
        password="your_password"
    )
    
    async with EarthExternalFitnessSDK(settings) as sdk:
        # 2. If no experiment_id is provided, interactively pick one from active runs
        if not settings.experiment_id:
            await sdk.select_experiment()
        
        # 3. Start the background infrastructure (heartbeats and polling)
        await sdk.start()
        
        # 4. Iterate over tasks and submit results as they become available
        async for individual in sdk.tasks():
            print(f"Calculating fitness for individual {individual.individual_id}")
            
            # Implementation of your computation logic
            result = FitnessCalculationResult(score=42.0)
            
            await sdk.submit_result(individual.individual_id, result)

if __name__ == "__main__":
    asyncio.run(main())

Advanced Usage: Parallelism with ProcessPoolExecutor

For CPU-intensive fitness calculations, use a ProcessPoolExecutor to utilize multiple cores while the SDK handles communication in the main async event loop.

import asyncio
import concurrent.futures
from earth_external_fitness import EarthExternalFitnessSDK, EarthSDKSettings, FitnessCalculationResult

def heavy_fitness_calc(chromosome):
    """
    CPU-bound calculation running in a separate process.
    """
    # ... perform complex scientific compute ...
    return 10.5  # Return the score

async def main():
    settings = EarthSDKSettings(
        gateway_url="https://earth.evolution.inc/api/external-fitness",
        experiment_id="your-experiment-uuid"
    )
    
    # Initialize a Process Pool for parallel compute
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as pool:
        loop = asyncio.get_running_loop()
        
        async with EarthExternalFitnessSDK(settings) as sdk:
            await sdk.start()
            
            tasks = set()
            
            async for individual in sdk.tasks():
                # Schedule the CPU-bound task in the process pool
                fut = loop.run_in_executor(pool, heavy_fitness_calc, individual.chromosome)
                
                # Wrap the submission in an async task to keep the loop moving
                async def handle_and_submit(ind_id, calculation_future):
                    try:
                        score = await calculation_future
                        result = FitnessCalculationResult(score=score)
                        await sdk.submit_result(ind_id, result)
                    except Exception as e:
                        # Report local failure back to the platform
                        await sdk.submit_result(ind_id, error=str(e))

                t = asyncio.create_task(handle_and_submit(individual.individual_id, fut))
                tasks.add(t)
                t.add_done_callback(tasks.discard)

            # Ensure all final results are submitted before exiting
            if tasks:
                await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

Configuration (Environment Variables)

Settings can be managed via environment variables prefixed with EARTH_:

Variable	Description	Default
EARTH_BASE_URL	Base API URL	https://earth.evolution.inc
EARTH_USERNAME	Platform username	None
EARTH_PASSWORD	Platform password	None
EARTH_EXPERIMENT_ID	Target Experiment UUID	None
EARTH_HEARTBEAT_INTERVAL	Seconds between pings	60