from google.adk.agents import Agent
# ── Mock Parts Inventory ──────────────────────────────────────────────────────
PARTS_INVENTORY = {
"Brake pads x4": {
"depot_a_manchester": 12,
"depot_b_birmingham": 4,
"depot_c_leeds": 0,
"unit_cost_gbp": 140,
"lead_time_days": 0,
},
"Brake adjustment kit": {
"depot_a_manchester": 8,
"depot_b_birmingham": 0,
"depot_c_leeds": 3,
"unit_cost_gbp": 35,
"lead_time_days": 0,
},
"Refrigerant R-404A": {
"depot_a_manchester": 2,
"depot_b_birmingham": 0,
"depot_c_leeds": 0,
"unit_cost_gbp": 320,
"lead_time_days": 1,
},
"Compressor filter kit": {
"depot_a_manchester": 0,
"depot_b_birmingham": 1,
"depot_c_leeds": 0,
"unit_cost_gbp": 85,
"lead_time_days": 2,
},
"Nitrogen inflation kit": {
"depot_a_manchester": 5,
"depot_b_birmingham": 5,
"depot_c_leeds": 2,
"unit_cost_gbp": 20,
"lead_time_days": 0,
},
"Tyre pressure gauge": {
"depot_a_manchester": 10,
"depot_b_birmingham": 8,
"depot_c_leeds": 6,
"unit_cost_gbp": 15,
"lead_time_days": 0,
},
}
# ── Mock Asset Database ───────────────────────────────────────────────────────
ASSET_DB = {
"TRL-001": {
"type": "Reefer Trailer",
"brake_wear_pct": 78,
"tyre_pressure_bar": 7.2,
"tru_fault_code": "TRU-E04",
"last_service_days_ago": 42,
"mileage_since_service": 12400,
},
"TRL-002": {
"type": "Curtainsider",
"brake_wear_pct": 31,
"tyre_pressure_bar": 8.1,
"tru_fault_code": None,
"last_service_days_ago": 12,
"mileage_since_service": 3200,
},
"TNK-001": {
"type": "Bulk Tanker",
"brake_wear_pct": 65,
"tyre_pressure_bar": 6.8,
"tru_fault_code": None,
"last_service_days_ago": 30,
"mileage_since_service": 9800,
},
}
# ── Parts Tools ───────────────────────────────────────────────────────────────
def check_parts_availability(part_name: str) -> dict:
"""
Checks depot stock levels for a specific part.
Args:
part_name: The name of the part to check
Returns:
Stock levels across all depots, unit cost, and lead time.
"""
matched = None
for key in PARTS_INVENTORY:
if part_name.lower() in key.lower() or key.lower() in part_name.lower():
matched = key
break
if not matched:
return {
"error": f"Part '{part_name}' not found.",
"available_parts": list(PARTS_INVENTORY.keys()),
}
stock = PARTS_INVENTORY[matched]
in_stock_depots = {
depot: qty
for depot, qty in stock.items()
if depot.startswith("depot") and qty > 0
}
return {
"part": matched,
"unit_cost_gbp": stock["unit_cost_gbp"],
"lead_time_days": stock["lead_time_days"],
"stock_by_depot": {k: v for k, v in stock.items() if k.startswith("depot")},
"available_at": in_stock_depots if in_stock_depots else "OUT OF STOCK at all depots",
"pre_position_signal": (
"SEND PRE-POSITION ORDER NOW — stock critically low, lead time > 0 days"
if stock["lead_time_days"] > 0 or not in_stock_depots
else "Stock sufficient — no pre-order needed"
),
}
def check_parts_for_job_card(parts_list: str) -> dict:
"""
Checks availability for all parts required by a job card at once.
Args:
parts_list: Comma-separated list of parts needed
Returns:
Consolidated availability report with total estimated cost.
"""
parts = [p.strip() for p in parts_list.split(",")]
results = []
total_cost = 0
has_shortage = False
for part in parts:
availability = check_parts_availability(part)
if "error" not in availability:
total_cost += availability["unit_cost_gbp"]
if (
availability["lead_time_days"] > 0
or availability["available_at"] == "OUT OF STOCK at all depots"
):
has_shortage = True
results.append(availability)
return {
"parts_checked": len(parts),
"total_estimated_cost_gbp": total_cost,
"shortage_alert": has_shortage,
"pre_position_required": has_shortage,
"parts_detail": results,
"recommendation": (
"PARTS PRE-POSITION ORDER REQUIRED — place order 48-72 hours before asset arrives."
if has_shortage
else "All parts in stock. Asset can be serviced immediately on arrival."
),
}
# ── Parts Sub-Agent ───────────────────────────────────────────────────────────
parts_agent = Agent(
name="parts_inventory_agent",
model="gemini-1.5-flash",
description="Specialist sub-agent for parts and inventory management across depots.",
instruction="""
You are the Parts and Inventory Agent for a trailer and tanker leasing company.
Check stock availability for parts across all depots.
Report unit costs and lead times clearly.
Flag parts that need pre-ordering prominently.
Use check_parts_for_job_card when checking multiple parts at once.
""",
tools=[check_parts_availability, check_parts_for_job_card],
)
# ── Orchestrator Tools ────────────────────────────────────────────────────────
def get_asset_telemetry(asset_id: str) -> dict:
"""
Fetches current telemetry data for a given asset.
Args:
asset_id: The unique asset identifier e.g. TRL-001
Returns:
Dictionary with current sensor readings for the asset.
"""
asset_id = asset_id.upper().strip()
if asset_id not in ASSET_DB:
return {
"error": f"Asset {asset_id} not found.",
"available_assets": list(ASSET_DB.keys()),
}
return {"asset_id": asset_id, **ASSET_DB[asset_id]}
def predict_failure_risk(asset_id: str) -> dict:
"""
Predicts component failure risk for a given asset.
Args:
asset_id: The unique asset identifier
Returns:
Risk assessment with component-level scores and overall priority.
"""
asset_id = asset_id.upper().strip()
if asset_id not in ASSET_DB:
return {"error": f"Asset {asset_id} not found."}
data = ASSET_DB[asset_id]
risks = []
if data["brake_wear_pct"] > 70:
risks.append({
"component": "Brake System",
"risk_level": "HIGH",
"estimated_days_to_failure": 3,
"recommended_action": "Immediate brake adjustment or pad replacement",
"parts_needed": "Brake pads x4, Brake adjustment kit",
})
elif data["brake_wear_pct"] > 50:
risks.append({
"component": "Brake System",
"risk_level": "MEDIUM",
"estimated_days_to_failure": 10,
"recommended_action": "Schedule brake inspection within 7 days",
"parts_needed": "Brake pads x4",
})
if data["tyre_pressure_bar"] < 7.0:
risks.append({
"component": "Tyres",
"risk_level": "HIGH",
"estimated_days_to_failure": 2,
"recommended_action": "Immediate tyre pressure check and inflation",
"parts_needed": "Nitrogen inflation kit, Tyre pressure gauge",
})
if data["tru_fault_code"]:
risks.append({
"component": "Refrigeration Unit (TRU)",
"risk_level": "HIGH",
"estimated_days_to_failure": 1,
"recommended_action": f"TRU fault {data['tru_fault_code']} detected — inspect compressor immediately",
"parts_needed": "Refrigerant R-404A, Compressor filter kit",
})
priority = "LOW"
if any(r["risk_level"] == "HIGH" for r in risks):
priority = "CRITICAL"
elif any(r["risk_level"] == "MEDIUM" for r in risks):
priority = "MODERATE"
all_parts = ", ".join(r["parts_needed"] for r in risks if r.get("parts_needed"))
return {
"asset_id": asset_id,
"asset_type": data["type"],
"overall_priority": priority,
"component_risks": risks if risks else [{
"component": "All Systems",
"risk_level": "LOW",
"recommended_action": "No immediate action required",
"parts_needed": "",
}],
"all_parts_needed": all_parts,
}
def generate_job_card(asset_id: str) -> dict:
"""
Auto-generates a structured repair job card for a given asset.
Args:
asset_id: The unique asset identifier
Returns:
Structured job card with jobs and parts required.
"""
risk = predict_failure_risk(asset_id)
if "error" in risk:
return risk
jobs = [
f"[{c['risk_level']}] {c['component']}: {c['recommended_action']}"
for c in risk["component_risks"]
if c["risk_level"] in ("HIGH", "MEDIUM")
]
return {
"job_card_id": f"JC-{asset_id}-AUTO",
"asset_id": asset_id,
"asset_type": risk["asset_type"],
"priority": risk["overall_priority"],
"jobs": jobs if jobs else ["No urgent jobs — routine inspection only"],
"parts_required": risk["all_parts_needed"] or "None",
"depot_routing": "Depot A — Manchester (nearest depot)",
"technician_skill": "Heavy Vehicle Technician Level 2",
"status": "PENDING PARTS CONFIRMATION",
}
def get_fleet_risk_summary() -> dict:
"""
Returns a ranked risk summary across all assets in the fleet.
Returns:
All assets ranked by overall priority with top risks highlighted.
"""
summary = []
for asset_id in ASSET_DB:
risk = predict_failure_risk(asset_id)
summary.append({
"asset_id": asset_id,
"asset_type": risk["asset_type"],
"priority": risk["overall_priority"],
"top_risk": risk["component_risks"][0]["component"] if risk["component_risks"] else "None",
"parts_needed": risk.get("all_parts_needed", "None"),
})
priority_order = {"CRITICAL": 0, "MODERATE": 1, "LOW": 2}
summary.sort(key=lambda x: priority_order.get(x["priority"], 3))
return {
"total_assets": len(summary),
"critical_count": sum(1 for a in summary if a["priority"] == "CRITICAL"),
"moderate_count": sum(1 for a in summary if a["priority"] == "MODERATE"),
"fleet_ranked": summary,
}
# ── Root Orchestrator Agent ───────────────────────────────────────────────────
root_agent = Agent(
name="fleetsense_orchestrator",
model="gemini-1.5-flash",
description="TCS FleetSense Orchestrator: coordinates asset health assessment and delegates parts queries to the Parts Inventory sub-agent.",
instruction="""
You are FleetSense, the AI maintenance orchestrator for a trailer and tanker leasing company.
You coordinate with a specialist Parts Inventory sub-agent for complete maintenance decisions.
Your workflow for ANY asset query:
1. Fetch telemetry with get_asset_telemetry
2. Predict failure risk with predict_failure_risk
3. If risk is MEDIUM or CRITICAL generate a job card with generate_job_card
4. Always delegate parts availability to the parts_inventory_agent sub-agent
5. Combine both results into a final recommendation for the technician
For fleet overview queries use get_fleet_risk_summary then check parts for CRITICAL assets.
Available assets: TRL-001 (Reefer Trailer), TRL-002 (Curtainsider), TNK-001 (Bulk Tanker)
Be concise and direct. Lead with priority and action.
""",
tools=[get_asset_telemetry, predict_failure_risk, generate_job_card, get_fleet_risk_summary],
sub_agents=[parts_agent],
)
