pyp6xer_earned_value
Calculate Earned Value Management metrics from Primavera P6 XER data. Provides BCWS, BCWP, ACWP, SPI, CPI, CV, SV, EAC, VAC.
Instructions
Calculate Earned Value Management (EVM) metrics.
Metrics:
BCWS (PV): Budgeted Cost of Work Scheduled = total budgeted cost × duration %
BCWP (EV): Budgeted Cost of Work Performed = sum of (budget × % complete) per task
ACWP (AC): Actual Cost of Work Performed = sum of actual costs
SPI: Schedule Performance Index = EV / PV
CPI: Cost Performance Index = EV / AC
CV: Cost Variance = EV - AC
SV: Schedule Variance = EV - PV
EAC: Estimate at Completion = BAC / CPI
VAC: Variance at Completion = BAC - EAC
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| cache_key | No | Cache key identifying the loaded XER file (set when calling pyp6xer_load_file) | default |
| proj_id | No | Project ID or short name; uses first project if omitted |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- server.py:1357-1411 (handler)The handler function that calculates Earned Value Management (EVM) metrics: BCWS (PV), BCWP (EV), ACWP (AC), SPI, CPI, CV, SV, EAC, VAC.
def pyp6xer_earned_value( cache_key: Annotated[str, Field(description="Cache key identifying the loaded XER file (set when calling pyp6xer_load_file)")] = "default", proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None, ctx: Context = None, ) -> str: """Calculate Earned Value Management (EVM) metrics. Metrics: - BCWS (PV): Budgeted Cost of Work Scheduled = total budgeted cost × duration % - BCWP (EV): Budgeted Cost of Work Performed = sum of (budget × % complete) per task - ACWP (AC): Actual Cost of Work Performed = sum of actual costs - SPI: Schedule Performance Index = EV / PV - CPI: Cost Performance Index = EV / AC - CV: Cost Variance = EV - AC - SV: Schedule Variance = EV - PV - EAC: Estimate at Completion = BAC / CPI - VAC: Variance at Completion = BAC - EAC """ xer = _get_xer(ctx, cache_key) proj = _get_project(xer, proj_id) tasks = proj.tasks if proj_id else list(xer.tasks.values()) bac = sum(t.budgeted_cost for t in tasks) # Budget at Completion acwp = sum(t.actual_cost for t in tasks) bcwp = sum(t.budgeted_cost * t.percent_complete for t in tasks) bcws = bac * proj.duration_percent # simplified PV spi = round(bcwp / bcws, 3) if bcws else None cpi = round(bcwp / acwp, 3) if acwp else None eac = round(bac / cpi, 2) if cpi else None vac = round(bac - eac, 2) if eac else None return json.dumps({ "data_date": _fmt_date(proj.data_date), "BAC": round(bac, 2), "BCWS_PV": round(bcws, 2), "BCWP_EV": round(bcwp, 2), "ACWP_AC": round(acwp, 2), "SPI": spi, "CPI": cpi, "CV": round(bcwp - acwp, 2), "SV": round(bcwp - bcws, 2), "EAC": eac, "VAC": vac, "interpretation": { "SPI": ( "On schedule" if spi and spi >= 1.0 else "Behind schedule" if spi else "N/A" ), "CPI": ( "Under budget" if cpi and cpi >= 1.0 else "Over budget" if cpi else "N/A" ), }, }, indent=2) - server.py:1356-1361 (schema)Tool registration with FastMCP decorator defining the tool name 'pyp6xer_earned_value' and its parameters (cache_key, proj_id, ctx with Pydantic annotations).
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False)) def pyp6xer_earned_value( cache_key: Annotated[str, Field(description="Cache key identifying the loaded XER file (set when calling pyp6xer_load_file)")] = "default", proj_id: Annotated[str | None, Field(description="Project ID or short name; uses first project if omitted")] = None, ctx: Context = None, ) -> str: - server.py:1355-1357 (registration)Registration via @mcp.tool decorator - registers the function as an MCP tool with read-only, non-destructive, idempotent, and open-world hints.
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False)) def pyp6xer_earned_value( - server.py:1490-1536 (helper)The pyp6xer_generate_report helper reuses the same earned value calculation logic (lines 1490-1495) for monthly report generation.
bac = sum(t.budgeted_cost for t in tasks) acwp = sum(t.actual_cost for t in tasks) bcwp = sum(t.budgeted_cost * t.percent_complete for t in tasks) # 0.0–1.0 scale bcws = bac * proj.duration_percent spi = round(bcwp / bcws, 3) if bcws else None cpi = round(bcwp / acwp, 3) if acwp else None return json.dumps({ "report_type": "monthly_progress", "project": { "short_name": proj.short_name, "name": proj.name, "data_date": _fmt_date(proj.data_date), "planned_finish": _fmt_date(proj.finish_date), }, "progress": { "total_activities": len(tasks), "not_started": not_started, "in_progress": in_progress, "completed": completed, "weighted_percent_complete": round(weighted_pct * 100, 1), "milestones_total": len(milestones), "milestones_complete": ms_done, }, "health": { "score": score, "rating": ( "Excellent" if score >= 85 else "Good" if score >= 70 else "Fair" if score >= 55 else "Poor" ), "issues": issues, }, "critical_path": { "count": len(critical), "pct_of_total": round(len(critical) / len(tasks) * 100, 1) if tasks else 0, }, "slipping_activities": slipping[:10], "slipping_count": len(slipping), "earned_value": { "BAC": round(bac, 2), "SPI": spi, "CPI": cpi, "ACWP": round(acwp, 2), "BCWP": round(bcwp, 2), }, "cost_summary": {