Metadata-Version: 2.4
Name: plansolve
Version: 0.25.2
Summary: Python SDK for PlanSolve optimization API
Author-email: PlanSolve <support@plansolve.app>
License: MIT
Project-URL: Homepage, https://plansolve.app
Project-URL: Documentation, https://plansolve.app/docs
Project-URL: Repository, https://github.com/plansolve/plansolve-python-sdk
Project-URL: Issues, https://github.com/plansolve/plansolve-python-sdk/issues
Keywords: optimization,routing,scheduling,field-service,professional-services
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.31.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Dynamic: license-file

# 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:

```bash
pip install plansolve
```

Or via pipenv:

```bash
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)

```python
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:

```python
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:

```python
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)

```python
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:

```python
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

```python
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

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

#### Shift

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

#### Visit

```python
{
    '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

```python
{
    '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.

```python
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

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

#### SolverOptions

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

### Professional Services Models

#### Employee

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

#### Task

```python
{
    '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

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

#### Task

```python
{
    '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:

```python
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

```python
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.
