Building a Pool Service Agent: Maintenance Scheduling, Chemical Balance, and Equipment Repair

The Pool Service Operations Model

Pool service companies run route-based businesses. A technician visits 8-12 pools per day, testing water chemistry, adding chemicals, cleaning filters, and inspecting equipment. The difference between a profitable pool service company and a struggling one often comes down to route efficiency and chemical accuracy. An AI agent that optimizes routes, calculates exact chemical dosages, diagnoses equipment problems before they become emergencies, and manages seasonal transitions can increase the number of pools each technician services by 20-30%.

Chemical balance is where the AI adds the most technical value. Pool chemistry involves multiple interacting variables — pH, alkalinity, calcium hardness, cyanuric acid, and sanitizer levels — where adjusting one affects the others.

Chemical Balance Calculator

Pool chemistry requires precise calculations based on pool volume, current readings, and target ranges. The agent calculates exact dosages.

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

from dataclasses import dataclass
from typing import Optional

@dataclass
class WaterTestResults:
    ph: float
    free_chlorine: float          # ppm
    total_alkalinity: float       # ppm
    calcium_hardness: float       # ppm
    cyanuric_acid: float          # ppm
    total_dissolved_solids: float  # ppm
    temperature_f: float
    pool_volume_gallons: int

TARGET_RANGES = {
    "ph": (7.2, 7.6),
    "free_chlorine": (1.0, 3.0),
    "total_alkalinity": (80, 120),
    "calcium_hardness": (200, 400),
    "cyanuric_acid": (30, 50),
}

class ChemicalCalculator:
    def calculate_adjustments(self, readings: WaterTestResults) -> list[dict]:
        adjustments = []
        volume = readings.pool_volume_gallons

        # pH adjustment
        if readings.ph < TARGET_RANGES["ph"][0]:
            deficit = TARGET_RANGES["ph"][0] - readings.ph
            soda_ash_oz = deficit * volume / 10000 * 6
            adjustments.append({
                "parameter": "pH (raise)",
                "current": readings.ph,
                "target": TARGET_RANGES["ph"][0],
                "chemical": "Soda Ash (sodium carbonate)",
                "amount_oz": round(soda_ash_oz, 1),
                "instruction": "Dissolve in bucket of water, pour along edges with pump running",
            })
        elif readings.ph > TARGET_RANGES["ph"][1]:
            excess = readings.ph - TARGET_RANGES["ph"][1]
            muriatic_oz = excess * volume / 10000 * 16
            adjustments.append({
                "parameter": "pH (lower)",
                "current": readings.ph,
                "target": TARGET_RANGES["ph"][1],
                "chemical": "Muriatic Acid (31.45%)",
                "amount_oz": round(muriatic_oz, 1),
                "instruction": "Add slowly to deep end with pump running. Retest in 4 hours.",
            })

        # Chlorine adjustment
        if readings.free_chlorine < TARGET_RANGES["free_chlorine"][0]:
            deficit = TARGET_RANGES["free_chlorine"][0] - readings.free_chlorine
            # Account for CYA stabilizer effect on effective chlorine
            cya_factor = max(1.0, readings.cyanuric_acid / 30)
            shock_oz = deficit * volume / 10000 * 2 * cya_factor
            adjustments.append({
                "parameter": "Free Chlorine (raise)",
                "current": readings.free_chlorine,
                "target": TARGET_RANGES["free_chlorine"][0],
                "chemical": "Calcium Hypochlorite (67%)",
                "amount_oz": round(shock_oz, 1),
                "instruction": "Pre-dissolve in bucket, add to pool at dusk for best results",
            })

        # Alkalinity adjustment
        if readings.total_alkalinity < TARGET_RANGES["total_alkalinity"][0]:
            deficit = TARGET_RANGES["total_alkalinity"][0] - readings.total_alkalinity
            bicarb_lbs = deficit * volume / 10000 * 1.4 / 16
            adjustments.append({
                "parameter": "Total Alkalinity (raise)",
                "current": readings.total_alkalinity,
                "target": TARGET_RANGES["total_alkalinity"][0],
                "chemical": "Sodium Bicarbonate (baking soda)",
                "amount_lbs": round(bicarb_lbs, 1),
                "instruction": "Broadcast across surface with pump running. Max 10 lbs per treatment.",
            })

        return adjustments

    def calculate_saturation_index(self, readings: WaterTestResults) -> dict:
        """Langelier Saturation Index: predicts scaling or corrosion tendency."""
        import math
        temp_factor = 0.0 + (readings.temperature_f - 32) * 0.01
        tf = round(temp_factor, 2)
        cf = round(math.log10(readings.calcium_hardness) - 0.4, 2)
        af = round(math.log10(readings.total_alkalinity), 2)
        lsi = readings.ph - (9.3 + tf + cf + af)
        lsi = round(lsi, 2)

        if lsi > 0.3:
            condition = "scaling"
            action = "Lower pH or calcium hardness to prevent scale buildup"
        elif lsi < -0.3:
            condition = "corrosive"
            action = "Raise pH or alkalinity to prevent equipment corrosion"
        else:
            condition = "balanced"
            action = "Water is balanced — no action needed"

        return {"lsi": lsi, "condition": condition, "action": action}

Service Route Optimization

Route efficiency directly impacts profitability. The agent optimizes the sequence of pool visits to minimize drive time.

from math import radians, sin, cos, sqrt, atan2

def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    R = 3959
    dlat = radians(lat2 - lat1)
    dlon = radians(lon2 - lon1)
    a = sin(dlat/2)**2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon/2)**2
    return R * 2 * atan2(sqrt(a), sqrt(1 - a))

class RouteOptimizer:
    def optimize_daily_route(
        self, start_location: tuple, pools: list[dict],
    ) -> list[dict]:
        """Nearest-neighbor heuristic for route optimization."""
        remaining = list(pools)
        route = []
        current_lat, current_lon = start_location

        while remaining:
            nearest = min(
                remaining,
                key=lambda p: haversine(current_lat, current_lon, p["lat"], p["lon"]),
            )
            distance = haversine(current_lat, current_lon, nearest["lat"], nearest["lon"])
            route.append({
                "stop": len(route) + 1,
                "address": nearest["address"],
                "customer": nearest["customer_name"],
                "distance_from_previous": round(distance, 1),
                "estimated_service_time_min": nearest.get("service_time", 30),
                "special_notes": nearest.get("notes", ""),
            })
            current_lat, current_lon = nearest["lat"], nearest["lon"]
            remaining.remove(nearest)

        total_distance = sum(s["distance_from_previous"] for s in route)
        total_time = sum(s["estimated_service_time_min"] for s in route)
        return {
            "stops": route,
            "total_distance_miles": round(total_distance, 1),
            "total_service_time_hours": round(total_time / 60, 1),
            "estimated_drive_time_hours": round(total_distance / 25, 1),
        }

Equipment Diagnostics

Pool equipment fails in predictable patterns. The agent diagnoses issues from symptoms and recommends repairs.

EQUIPMENT_DIAGNOSTICS = {
    "pump_not_priming": {
        "symptoms": ["pump running but no water flow", "air bubbles in pump basket"],
        "probable_causes": [
            {"cause": "Air leak in suction line", "likelihood": "high",
             "fix": "Check and replace O-rings on pump lid and unions", "cost_range": "$15-45"},
            {"cause": "Clogged impeller", "likelihood": "medium",
             "fix": "Remove pump housing and clear debris from impeller", "cost_range": "$85-150"},
            {"cause": "Low water level", "likelihood": "high",
             "fix": "Fill pool to mid-skimmer level", "cost_range": "$0"},
        ],
    },
    "heater_not_firing": {
        "symptoms": ["heater turns on but no heat", "error codes on display"],
        "probable_causes": [
            {"cause": "Dirty or failed pressure switch", "likelihood": "high",
             "fix": "Clean or replace pressure switch", "cost_range": "$45-120"},
            {"cause": "Failed ignitor", "likelihood": "medium",
             "fix": "Replace hot surface ignitor", "cost_range": "$80-200"},
            {"cause": "Low gas pressure", "likelihood": "low",
             "fix": "Contact gas company to check supply pressure", "cost_range": "$0"},
        ],
    },
    "filter_pressure_high": {
        "symptoms": ["pressure gauge above 25 PSI", "reduced water flow"],
        "probable_causes": [
            {"cause": "Dirty filter cartridge or grids", "likelihood": "high",
             "fix": "Clean or replace filter media", "cost_range": "$0-300"},
            {"cause": "Clogged return lines", "likelihood": "low",
             "fix": "Professional line cleaning required", "cost_range": "$150-350"},
        ],
    },
}

def diagnose_equipment(symptom_description: str) -> dict:
    description_lower = symptom_description.lower()
    for issue_key, issue in EQUIPMENT_DIAGNOSTICS.items():
        for symptom in issue["symptoms"]:
            if any(word in description_lower for word in symptom.split()):
                return {
                    "issue": issue_key.replace("_", " ").title(),
                    "matching_symptoms": issue["symptoms"],
                    "probable_causes": issue["probable_causes"],
                    "recommendation": issue["probable_causes"][0]["fix"],
                    "estimated_cost": issue["probable_causes"][0]["cost_range"],
                }
    return {
        "issue": "Unknown",
        "recommendation": "Schedule on-site diagnostic visit",
        "estimated_cost": "$95 diagnostic fee",
    }

Seasonal Planning

Pool services have distinct seasonal phases. The agent manages transitions and prepares for each season.

class SeasonalPlanner:
    SEASONAL_TASKS = {
        "spring_opening": [
            {"task": "Remove cover and clean", "order": 1, "time_min": 30},
            {"task": "Inspect equipment (pump, filter, heater)", "order": 2, "time_min": 20},
            {"task": "Fill to operating level", "order": 3, "time_min": 15},
            {"task": "Prime and start pump", "order": 4, "time_min": 10},
            {"task": "Initial chemical treatment (shock)", "order": 5, "time_min": 15},
            {"task": "Install ladders and accessories", "order": 6, "time_min": 15},
        ],
        "fall_closing": [
            {"task": "Lower water level below returns", "order": 1, "time_min": 20},
            {"task": "Blow out plumbing lines", "order": 2, "time_min": 30},
            {"task": "Add winterizing chemicals", "order": 3, "time_min": 10},
            {"task": "Install winter plugs", "order": 4, "time_min": 15},
            {"task": "Install pool cover", "order": 5, "time_min": 30},
            {"task": "Disconnect and store pump/filter", "order": 6, "time_min": 20},
        ],
    }

    def generate_seasonal_schedule(
        self, pools: list[dict], season: str, start_date: str,
    ) -> list[dict]:
        tasks = self.SEASONAL_TASKS.get(season, [])
        total_time_per_pool = sum(t["time_min"] for t in tasks)
        pools_per_day = max(1, int(480 / total_time_per_pool))  # 8-hour day

        schedule = []
        for i, pool in enumerate(pools):
            day_offset = i // pools_per_day
            schedule.append({
                "customer": pool["customer_name"],
                "address": pool["address"],
                "scheduled_day": f"Day {day_offset + 1}",
                "tasks": [t["task"] for t in tasks],
                "estimated_time_min": total_time_per_pool,
            })
        return schedule

FAQ

How does the agent account for different pool types in chemical calculations?

The calculations adjust based on pool type (chlorine, saltwater, biguanide) and surface material (plaster, fiberglass, vinyl). Saltwater pools require different alkalinity targets and do not need external chlorine unless the salt cell is underperforming. The agent stores the pool type in the customer profile and applies the correct formula set automatically.

Can the agent predict when equipment will fail?

Yes, through trend analysis. The agent tracks filter pressure readings, pump amperage, and heater cycle counts over time. When pressure rises steadily between cleanings, it indicates filter media degradation. When pump amperage increases, it signals bearing wear. The agent flags these trends 2-4 weeks before likely failure, allowing proactive replacement during scheduled visits.

How does route optimization handle pools with different service frequencies?

Some pools are serviced weekly, others bi-weekly. The agent builds separate route sets for each frequency tier. On weeks when bi-weekly pools are due, it merges them into the weekly route using geographic clustering. This prevents the technician from driving past a bi-weekly pool on the way to a weekly one without stopping.

---

#PoolService #ChemicalBalance #ServiceRoutes #EquipmentDiagnostics #SeasonalPlanning #AgenticAI #LearnAI #AIEngineering