plansolve 0.25.6

Python SDK for PlanSolve optimization API

PlanSolve Python SDK

A Python client library for the PlanSolve optimization API. Solve complex field service routing, professional services task assignment, and shift scheduling problems with a simple, async API.

Installation

Install via pip:

pip install plansolve

Or via pipenv:

pipenv install plansolve

Requirements

• Python 3.8+
• PlanSolve API key (optional but recommended for production use)

API Workflow

All PlanSolve optimization APIs follow the same workflow:

1. Construct Request - Build your optimization request with the required data
2. Start Solver - Submit the request to start the optimization process
3. Poll Status - Check the status until the solver completes
4. Get Results - Retrieve the optimized solution

Quick Start

The SDK supports both dictionary-based requests (for flexibility) and dataclass-based requests (for type safety and IDE support). We recommend using dataclasses for better developer experience.

Field Service Optimization

Using Dataclasses (Recommended)

from plansolve import (
    PlanSolveClient,
    Vehicle,
    Visit,
    Shift,
    TimeWindow,
    FieldServiceRequest,
    ConstraintWeight,
    Weights,
    SolverOptions,
)

plan_solve = PlanSolveClient(api_key='YOUR_API_KEY')

# Create vehicles using dataclasses
vehicles = [
    Vehicle(
        id='tech1',
        location=(40.7128, -74.0060),  # NYC coordinates [lat, lng]
        shifts=[
            Shift(
                id='morning-shift',
                min_start_time='2024-01-15T08:00:00',
                max_end_time='2024-01-15T17:00:00'
            )
        ],
        skills=['repair', 'installation']
    ),
    Vehicle(
        id='tech2',
        location=(40.7128, -74.0060),
        shifts=[
            Shift(
                id='afternoon-shift',
                min_start_time='2024-01-15T10:00:00',
                max_end_time='2024-01-15T19:00:00'
            )
        ],
        skills=['repair']
    )
]

# Create visits using dataclasses
visits = [
    Visit(
        id='visit1',
        name='Times Square Repair',
        location=(40.7589, -73.9851),  # Times Square coordinates [lat, lng]
        time_windows=[
            TimeWindow(
                min_start_time='2024-01-15T09:00:00',
                max_end_time='2024-01-15T17:00:00'
            )
        ],
        service_duration='PT30M',  # ISO 8601 duration (30 minutes)
        priority='HIGH',
        required_skills=['repair']
    ),
    Visit(
        id='visit2',
        name='Penn Station Installation',
        location=(40.7505, -73.9934),  # Penn Station coordinates [lat, lng]
        time_windows=[
            TimeWindow(
                min_start_time='2024-01-15T10:00:00',
                max_end_time='2024-01-15T16:00:00'
            )
        ],
        service_duration='PT45M',  # 45 minutes
        priority='MEDIUM',
        required_skills=['installation']
    )
]

# Optional: Configure constraint weights
weights = Weights(
    prefer_high_priority=ConstraintWeight.create_hard(1),
    minimize_travel_time=ConstraintWeight.create_soft(1),
)

# Optional: Configure solver options
options = SolverOptions(spent_limit='PT5M')  # 5 minutes

# Create the optimization request using dataclass
request = FieldServiceRequest(
    vehicles=vehicles,
    visits=visits,
    weights=weights,
    options=options
)

try:
    # Start the optimization
    response = plan_solve.field_service.start(request)
    job_id = response.job_id
    print(f"Optimization started! Job ID: {job_id}")
    
    # Check status until completion
    import time
    while True:
        time.sleep(2)  # Wait 2 seconds
        status = plan_solve.field_service.get_status(job_id)
        print(f"Status: {status.get('status', 'UNKNOWN')}")
        
        if status.get('status') != 'RUNNING' and not status.get('solving', False):
            break
    
    if status.get('status') == 'COMPLETED':
        # Get the final result (returns dataclass)
        result = plan_solve.field_service.get_result(job_id)
        print('Optimization completed!')
        
        # Print the optimized routes
        for vehicle in result.vehicles:
            print(f"\nRoute for {vehicle.id}:")
            for visit_id in vehicle.visits:
                visit = next((v for v in result.visits if v.id == visit_id), None)
                if visit:
                    print(f"  - {visit.name}: {visit.arrival_time} - {visit.departure_time}")
    else:
        print(f"Optimization failed with status: {status.get('status')}")
except Exception as error:
    print(f"Optimization failed: {error}")

Using Dictionaries (Alternative)

You can also use plain dictionaries if you prefer:

from plansolve import PlanSolveClient

plan_solve = PlanSolveClient(api_key='YOUR_API_KEY')

# Create vehicles (field technicians)
vehicles = [
    {
        'id': 'tech1',
        'location': [40.7128, -74.0060],  # NYC coordinates [lat, lng]
        'shifts': [
            {
                'id': 'morning-shift',
                'minStartTime': '2024-01-15T08:00:00',
                'maxEndTime': '2024-01-15T17:00:00'
            }
        ],
        'skills': ['repair', 'installation']
    },
    # ... more vehicles
]

# Create visits (customer appointments)
visits = [
    {
        'id': 'visit1',
        'name': 'Times Square Repair',
        'location': [40.7589, -73.9851],
        'timeWindows': [
            {
                'minStartTime': '2024-01-15T09:00:00',
                'maxEndTime': '2024-01-15T17:00:00'
            }
        ],
        'serviceDuration': 'PT30M',
        'priority': 'HIGH',
        'requiredSkills': ['repair']
    },
    # ... more visits
]

# Create the optimization request
request = {
    'vehicles': vehicles,
    'visits': visits
}

try:
    # Start the optimization
    response = plan_solve.field_service.start(request)
    job_id = response['jobId'] if isinstance(response, dict) else response.job_id
    print(f"Optimization started! Job ID: {job_id}")
    
    # Use the convenience method to wait for completion
    result = plan_solve.field_service.start_and_wait_for_completion(request)
    print('Optimization completed!')
except Exception as error:
    print(f"Optimization failed: {error}")

Professional Services Optimization

Using Dataclasses (Recommended)

For professional services task assignment optimization with type-safe dataclasses:

from plansolve import PlanSolveClient, Employee, Task, Shift, ProfessionalServicesRequest

plan_solve = PlanSolveClient(api_key='YOUR_API_KEY')

# Create employees (consultants, developers, etc.)
employees = [
    Employee(
        id='emp1',
        shifts=[
            Shift(
                id='shift1',
                min_start_time='2024-01-15T08:00:00',
                max_end_time='2024-01-15T18:00:00'
            )
        ],
        skills=['Java', 'Spring', 'Kotlin']
    ),
    Employee(
        id='emp2',
        shifts=[
            Shift(
                id='shift2',
                min_start_time='2024-01-15T09:00:00',
                max_end_time='2024-01-15T17:00:00'
            )
        ],
        skills=['Python', 'SQL', 'React']
    )
]

# Create tasks (projects, assignments, etc.)
tasks = [
    Task(
        id='task1',
        name='Develop REST API',
        deadline='2024-01-20T17:00:00',
        duration='PT16H',  # ISO 8601 duration (16 hours)
        priority='HIGH',
        required_skills=['Java', 'Spring']
    ),
    Task(
        id='task2',
        name='Database Design',
        deadline='2024-01-22T17:00:00',
        duration='PT8H',  # 8 hours
        priority='MEDIUM',
        required_skills=['SQL', 'Database Design']
    )
]

# Create the optimization request
request = ProfessionalServicesRequest(
    employees=employees,
    tasks=tasks
)

try:
    # Option 1: Start and poll manually
    response = plan_solve.professional_services.start(request)
    job_id = response.job_id
    print(f"Optimization started! Job ID: {job_id}")
    
    status = None
    while True:
        import time
        time.sleep(2)  # Wait 2 seconds
        status = plan_solve.get_status(job_id)
        print(f"Status: {status['status']}")
        
        if not status.get('solving', False):
            break
    
    if status['status'] == 'COMPLETED':
        # Get the final result
        result = plan_solve.professional_services.get_result(job_id)
        print('Optimization completed!')
        
        # Print the optimized task assignments
        for employee in result.employees:
            print(f"\nEmployee {employee.id} assigned tasks:")
            for task_id in employee.tasks:
                task = next((t for t in result.tasks if t.id == task_id), None)
                if task:
                    print(f"  - {task.name}: {task.start_time} - {task.end_time}")
    
    # Option 2: Use the convenience method to wait for completion
    result = plan_solve.professional_services.start_and_wait_for_completion(request)
    print('Optimization completed with result!')
    
except Exception as error:
    print(f"Optimization failed: {error}")

Using Dictionaries (Alternative)

from plansolve import PlanSolveClient

plan_solve = PlanSolveClient(api_key='YOUR_API_KEY')

# Create employees
employees = [
    {
        'id': 'emp1',
        'shifts': [
            {
                'id': 'shift1',
                'minStartTime': '2024-01-15T08:00:00',
                'maxEndTime': '2024-01-15T18:00:00'
            }
        ],
        'skills': ['Java', 'Spring', 'Kotlin']
    },
    # ... more employees
]

# Create tasks
tasks = [
    {
        'id': 'task1',
        'name': 'Develop REST API',
        'deadline': '2024-01-20T17:00:00',
        'duration': 'PT16H',
        'priority': 'HIGH',
        'requiredSkills': ['Java', 'Spring']
    },
    # ... more tasks
]

# Create the optimization request
request = {
    'employees': employees,
    'tasks': tasks
}

try:
    response = plan_solve.professional_services.start(request)
    result = plan_solve.professional_services.start_and_wait_for_completion(request)
    print('Optimization completed!')
except Exception as error:
    print(f"Optimization failed: {error}")

Shift Assignment Optimization

Using Dataclasses (Recommended)

For shift scheduling and assignment:

from plansolve import PlanSolveClient
from plansolve.shift import Employee, Task, Shift, ShiftRequest

plan_solve = PlanSolveClient(api_key='YOUR_API_KEY')

# Create employees
employees = [
    Employee(
        id='emp1',
        shifts=[
            Shift(
                id='shift1',
                min_start_time='2024-01-15T08:00:00',
                max_end_time='2024-01-15T18:00:00'
            )
        ],
        skills=['cashier', 'inventory']
    ),
    Employee(
        id='emp2',
        shifts=[
            Shift(
                id='shift2',
                min_start_time='2024-01-15T09:00:00',
                max_end_time='2024-01-15T17:00:00'
            )
        ],
        skills=['manager', 'cashier']
    )
]

# Create tasks (shift assignments)
tasks = [
    Task(
        id='task1',
        name='Morning Cashier Shift',
        deadline='2024-01-15T12:00:00',
        duration='PT4H',  # 4 hours
        priority='HIGH',
        required_skills=['cashier']
    ),
    Task(
        id='task2',
        name='Inventory Management',
        deadline='2024-01-15T16:00:00',
        duration='PT2H',  # 2 hours
        priority='MEDIUM',
        required_skills=['inventory']
    )
]

# Create the optimization request
request = ShiftRequest(
    employees=employees,
    tasks=tasks
)

try:
    # Option 1: Start and poll manually
    response = plan_solve.shift.start(request)
    job_id = response.job_id
    print(f"Optimization started! Job ID: {job_id}")
    
    status = None
    while True:
        import time
        time.sleep(2)
        status = plan_solve.shift.get_status(job_id)
        print(f"Status: {status['status']}")
        
        if not status.get('solving', False):
            break
    
    if status['status'] == 'COMPLETED':
        result = plan_solve.shift.get_result(job_id)
        print('Optimization completed!')
    
    # Option 2: Use the convenience method to wait for completion
    result = plan_solve.shift.start_and_wait_for_completion(request)
    print('Optimization completed with result!')
    
except Exception as error:
    print(f"Optimization failed: {error}")

API Reference

PlanSolveClient

The main client class for interacting with the PlanSolve API.

Constructor

PlanSolveClient(api_key: Optional[str] = None)

• api_key (optional): Your PlanSolve API key for authenticated requests

Methods

• field_service: Access to field service optimization methods
• professional_services: Access to professional services optimization methods
• shift: Access to shift assignment optimization methods
• get_status(job_id: str): Get the status of a running optimization job

FieldServiceApiClient

Handles field service optimization requests and results.

Methods

• start(request: Union[FieldServiceRequest, Dict]): Start a new field service optimization
• get_result(job_id: str): Get the completed optimization result
• get_status(job_id: str): Get the status of a running optimization job
• start_and_wait_for_completion(request, poll_interval_ms=5000, max_attempts=10): Start optimization and wait for completion with polling
• wait_for_completion(job_id, poll_interval_ms=5000, max_attempts=10): Wait for an existing job to complete

ProfessionalServicesApiClient

Handles professional services task assignment optimization requests and results.

Methods

• start(request: Union[ProfessionalServicesRequest, Dict]): Start a new professional services optimization
• get_result(job_id: str): Get the completed optimization result
• get_status(job_id: str): Get the status of a running optimization job
• start_and_wait_for_completion(request, poll_interval_ms=5000, max_attempts=1000): Start optimization and wait for completion with polling
• wait_for_completion(job_id, poll_interval_ms=5000, max_attempts=1000): Wait for an existing job to complete

ShiftApiClient

Handles shift assignment optimization requests and results.

Methods

• start(request: Union[ShiftRequest, Dict]): Start a new shift assignment optimization
• get_result(job_id: str): Get the completed optimization result
• get_status(job_id: str): Get the status of a running optimization job
• start_and_wait_for_completion(request, poll_interval_ms=5000, max_attempts=1000): Start optimization and wait for completion with polling
• wait_for_completion(job_id, poll_interval_ms=5000, max_attempts=1000): Wait for an existing job to complete

Data Models

Field Service Models

Vehicle

{
    'id': str,
    'location': [float, float],  # [latitude, longitude]
    'shifts': List[Shift],
    'skills': List[str]
}

Shift

{
    'id': str,
    'minStartTime': str,  # ISO 8601 datetime string
    'maxEndTime': str     # ISO 8601 datetime string
}

Visit

{
    'id': str,
    'name': str,
    'location': [float, float],  # [latitude, longitude]
    'timeWindows': List[TimeWindow],
    'serviceDuration': str,      # ISO 8601 duration string
    'priority': 'HIGH' | 'MEDIUM' | 'LOW',
    'requiredSkills': List[str]
}

TimeWindow

{
    'minStartTime': str,  # ISO 8601 datetime string
    'maxEndTime': str     # ISO 8601 datetime string
}

ConstraintWeight

The ConstraintWeight class represents a constraint weight in the format "Xhard/Ymedium/Zsoft" where X, Y, Z are numeric values. Only one of hard, medium, or soft can be non-zero at a time.

from plansolve import ConstraintWeight

# Create a hard constraint weight
hard_weight = ConstraintWeight.create_hard(1)

# Create a medium constraint weight
medium_weight = ConstraintWeight.create_medium(1)

# Create a soft constraint weight
soft_weight = ConstraintWeight.create_soft(1)

# Parse from string
weight = ConstraintWeight.parse("1hard/0medium/0soft")

# Convert to string
weight_str = str(weight)  # "1hard/0medium/0soft"

Weights

{
    'pinnedVisitVehicleAssignment': ConstraintWeight,
    'pinnedVisitServiceTime': ConstraintWeight,
    'noMissingSkills': ConstraintWeight,
    'serviceTimeMissing': ConstraintWeight,
    'vehicleUnassigned': ConstraintWeight,
    'preferHighPriority': ConstraintWeight,
    'minimizeTravelTime': ConstraintWeight,
    'preferUsingIdleVehicles': ConstraintWeight,
    'minimizeEarlyArrivalWait': ConstraintWeight,
    'preferInitialVehicleAssignment': ConstraintWeight,
    'PreferEarlierVisitDates': ConstraintWeight
}

SolverOptions

{
    'spentLimit': str,              # ISO-8601 duration format, e.g., "PT5M" for 5 minutes
    'unimprovedSpentLimit': str     # ISO-8601 duration format
}

Professional Services Models

Employee

{
    'id': str,
    'shifts': List[Shift],
    'skills': List[str]
}

Task

{
    'id': str,
    'name': str,
    'deadline': Optional[str],      # ISO 8601 datetime string
    'duration': str,                # ISO 8601 duration string
    'priority': str,
    'requiredSkills': List[str]
}

Shift Models

Employee

{
    'id': str,
    'shifts': List[Shift],
    'skills': List[str]
}

Task

{
    'id': str,
    'name': str,
    'deadline': Optional[str],      # ISO 8601 datetime string
    'duration': str,                # ISO 8601 duration string
    'priority': str,
    'requiredSkills': List[str]
}

Advanced Usage

Field Service vs Professional Services vs Shift

Field Service Optimization is designed for mobile workforce management:

• Optimizes vehicle routing with geographic constraints
• Considers travel time between locations
• Uses bounding box coordinates for optimization area
• Focuses on visit sequencing and route optimization

Professional Services Optimization is designed for task assignment:

• Assigns tasks to employees based on skills and availability
• No geographic constraints or travel time
• Considers employee shifts, skills, and task priorities
• Focuses on workload balancing and deadline management

Shift Assignment Optimization is designed for shift scheduling:

• Assigns shifts/tasks to employees based on skills and availability
• Similar to professional services but optimized for shift-based work
• Considers employee shifts, skills, and task priorities
• Focuses on shift coverage and scheduling

Asynchronous Optimization with Polling

For long-running optimizations, use the polling pattern:

import time

def wait_for_completion(client, job_id, max_wait_minutes=30):
    start_time = time.time()
    max_wait_seconds = max_wait_minutes * 60
    
    while time.time() - start_time < max_wait_seconds:
        status = client.get_status(job_id)
        
        if status['status'] == 'COMPLETED':
            return client.field_service.get_result(job_id)
        elif status['status'] == 'FAILED':
            raise Exception(f"Optimization failed: {status.get('error', 'Unknown error')}")
        
        # Wait before next poll
        time.sleep(5)
    
    raise Exception('Optimization timed out')

# Usage
result = wait_for_completion(plan_solve, response['jobId'])

Error Handling

try:
    response = plan_solve.field_service.start(request)
    # Handle success
except Exception as error:
    error_str = str(error)
    if '401' in error_str:
        print('Invalid API key')
    elif '400' in error_str:
        print('Invalid request data')
    elif '429' in error_str:
        print('Rate limit exceeded')
    else:
        print(f'Unexpected error: {error_str}')

Examples

See the examples/ directory for additional usage examples with both dictionary and dataclass approaches.

Support

For API support and questions, please refer to the main PlanSolve documentation or contact support.

License

This project is licensed under the MIT License - see the LICENSE file for details.