Optimizing technician routes for multi-site maintenance windows

import logging
import hashlib
import json
import time
from datetime import datetime, timedelta, timezone
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from enum import Enum

# --- Error Categorization ---
class RoutingErrorType(Enum):
    CONSTRAINT_VIOLATION = "CONSTRAINT_VIOLATION"
    WEATHER_EXCLUSION = "WEATHER_EXCLUSION"
    LEASE_WINDOW_MISMATCH = "LEASE_WINDOW_MISMATCH"
    DISPATCH_TIMEOUT = "DISPATCH_TIMEOUT"
    UNKNOWN = "UNKNOWN"

class RoutingException(Exception):
    def __init__(self, error_type: RoutingErrorType, message: str, site_id: Optional[str] = None):
        self.error_type = error_type
        self.site_id = site_id
        super().__init__(f"[{error_type.value}] {message}")

# --- Structured Logging Configuration ---
class ComplianceFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName
        }
        if hasattr(record, "site_id"):
            log_entry["site_id"] = record.site_id
        if hasattr(record, "error_type"):
            log_entry["error_type"] = record.error_type
        return json.dumps(log_entry)

logger = logging.getLogger("route_optimizer")
handler = logging.StreamHandler()
handler.setFormatter(ComplianceFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# --- Data Models ---
@dataclass
class SiteConstraint:
    site_id: str
    entry_window_start: datetime
    entry_window_end: datetime
    municipal_curfew_start: Optional[datetime] = None
    max_service_time_minutes: int = 60

@dataclass
class RouteSegment:
    from_site: str
    to_site: str
    travel_time_minutes: float
    arrival_time: datetime
    departure_time: datetime
    compliance_hash: str = ""

# --- Core Logic ---
def generate_compliance_hash(segment: RouteSegment) -> str:
    """Generate immutable SHA-256 hash for audit trail verification."""
    payload = json.dumps({
        "from": segment.from_site,
        "to": segment.to_site,
        "arrival": segment.arrival_time.isoformat(),
        "departure": segment.departure_time.isoformat()
    }, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

def validate_time_window(site: SiteConstraint, proposed_arrival: datetime) -> bool:
    """Enforce lease windows and municipal curfew hard stops."""
    if not (site.entry_window_start <= proposed_arrival <= site.entry_window_end):
        return False
    if site.municipal_curfew_start and proposed_arrival >= site.municipal_curfew_start:
        return False
    return True

def evaluate_weather_exclusion(site_id: str, forecast_data: Dict) -> bool:
    """Interface hook for Weather Window Optimization modules."""
    wind_speed_mph = forecast_data.get("wind_speed_mph", 0)
    lightning_prob = forecast_data.get("lightning_probability_pct", 0)
    # OSHA high-wind hard stop (~40 mph) for working at height; stricter
    # lease-specific caps are enforced upstream in Weather Window Optimization.
    if wind_speed_mph > 40 or lightning_prob > 15:
        logger.warning(
            "Site excluded due to unsafe meteorological conditions",
            extra={"site_id": site_id, "error_type": RoutingErrorType.WEATHER_EXCLUSION.value}
        )
        return True
    return False

def optimize_route_sequence(
    sites: List[SiteConstraint],
    start_location: str,
    start_time: datetime,
    travel_matrix: Dict[Tuple[str, str], float],
    weather_forecasts: Dict[str, Dict]
) -> List[RouteSegment]:
    """Deterministic TDVRPTW solver with compliance pre-filtering."""
    route = []
    current_time = start_time
    current_site = start_location

    for site in sorted(sites, key=lambda s: s.entry_window_start):
        if evaluate_weather_exclusion(site.site_id, weather_forecasts.get(site.site_id, {})):
            continue

        travel_time = travel_matrix.get((current_site, site.site_id), 60.0)
        arrival = current_time + timedelta(minutes=travel_time)

        if not validate_time_window(site, arrival):
            logger.warning(
                "Site excluded due to temporal constraint violation",
                extra={"site_id": site.site_id, "error_type": RoutingErrorType.CONSTRAINT_VIOLATION.value}
            )
            continue

        departure = arrival + timedelta(minutes=site.max_service_time_minutes)
        segment = RouteSegment(
            from_site=current_site,
            to_site=site.site_id,
            travel_time_minutes=travel_time,
            arrival_time=arrival,
            departure_time=departure
        )
        segment.compliance_hash = generate_compliance_hash(segment)
        route.append(segment)

        current_time = departure
        current_site = site.site_id

    return route

def dispatch_with_retry(route: List[RouteSegment], max_retries: int = 3) -> bool:
    """Resilient dispatch transmission with exponential backoff."""
    base_delay = 0.5
    for attempt in range(max_retries):
        try:
            # Simulate external dispatch API call
            time.sleep(base_delay * (2 ** attempt))
            logger.info("Dispatch payload transmitted successfully", extra={"route_segments": len(route)})
            return True
        except Exception as e:
            logger.error(
                f"Dispatch attempt {attempt + 1} failed",
                extra={"error_type": RoutingErrorType.DISPATCH_TIMEOUT.value, "detail": str(e)}
            )
            if attempt == max_retries - 1:
                raise RoutingException(RoutingErrorType.DISPATCH_TIMEOUT, "Max retries exceeded for dispatch")
    return False