21aecff0cb
Stage 11 backend (services/team_shape.py + api/team_shape.py):
- Ports V1's calculate_team_shape with the bug-4 fix already applied:
total_hours / manual_override on RatecardLine are per-1-asset; the
aggregator multiplies by line.volume.
- Programme roles never see efficiency cuts. Per-discipline efficiency
is capped at MAX_EFFICIENCY (90%).
- GET /opportunities/{id}/team-shape supports two modes:
* `efficiency_pct=N` blanket reduction across delivery roles, OR
* `discipline_overrides=<JSON>` per-discipline rates (Stage 10
efficiency profile output).
- Smoke-tested: 5,620.5 hrs / 3.12 FTE on the Versuni opportunity;
applying efficiency_pct=50 correctly halves to 2,810.25 hrs / 1.56 FTE.
Stage 11 frontend (Stage11TeamShape.tsx + api/teamShape.ts):
- Stats card (total / delivery / programme / adjusted FTE).
- Blanket efficiency slider 0–90% (disabled when per-discipline mode is
active so the two modes don't fight).
- One slider card per discipline showing live percentage; clearing all
reverts to the blanket slider.
- Per-discipline FTE table with Total hrs / FTE / Eff% / Adjusted hrs /
Adjusted FTE columns. Programme roles tagged with a badge.
Stage 6/7/8 backend tests (parallel test agent, +27 tests):
- test_assets.py (10): CRUD + sort_order auto-increment + cascade
delete + 400/404 paths + skipped real-Claude normalize.
- test_matching.py (10): 400 on no-assets, GET shape + ordering,
selection toggle deselects siblings, deselect leaves siblings alone,
cross-opp 404, skipped real-Claude run.
- test_ratecard.py (7): 400/404 paths, end-to-end build+get with the
bug-4 invariant explicitly asserted (line.base_hours == total_hours;
summary.total_hours == sum(base × volume)), no-selection skip,
rebuild idempotence, two-asset volume aggregation splits 1:2.
Suite: 100 collected, 95 passed, 5 skipped (all real-Anthropic), 0 failed.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
"""Stage 11 — team shape (FTE per role) endpoint."""
|
|
|
|
import json
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.models.opportunity import Opportunity
|
|
from app.services.team_shape import calculate_team_shape, HOURS_PER_FTE
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
async def _get_opp(db: AsyncSession, opportunity_id: int) -> Opportunity:
|
|
result = await db.execute(select(Opportunity).where(Opportunity.id == opportunity_id))
|
|
opp = result.scalar_one_or_none()
|
|
if opp is None:
|
|
raise HTTPException(status_code=404, detail=f"Opportunity {opportunity_id} not found")
|
|
return opp
|
|
|
|
|
|
@router.get("/{opportunity_id}/team-shape")
|
|
async def get_team_shape(
|
|
opportunity_id: int,
|
|
efficiency_pct: float = 0.0,
|
|
discipline_overrides: str = "",
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Compute FTE per role from the ratecard.
|
|
|
|
Query params:
|
|
- `efficiency_pct`: blanket percentage applied to every delivery role
|
|
(programme roles excluded).
|
|
- `discipline_overrides`: JSON-encoded `{discipline: pct}` map. When
|
|
provided, takes precedence over `efficiency_pct`.
|
|
|
|
Programme roles never see efficiency cuts. Per-discipline efficiency
|
|
is capped at MAX_EFFICIENCY (90%).
|
|
"""
|
|
opp = await _get_opp(db, opportunity_id)
|
|
|
|
overrides: dict[str, float] | None = None
|
|
if discipline_overrides:
|
|
try:
|
|
parsed = json.loads(discipline_overrides)
|
|
if isinstance(parsed, dict):
|
|
overrides = {str(k): float(v) for k, v in parsed.items()}
|
|
except (json.JSONDecodeError, TypeError, ValueError):
|
|
raise HTTPException(status_code=400, detail="discipline_overrides must be a JSON object of {discipline: percent}")
|
|
|
|
team = await calculate_team_shape(
|
|
db, opp,
|
|
efficiency_pct=efficiency_pct,
|
|
discipline_overrides=overrides,
|
|
)
|
|
|
|
total_hours = sum(t["total_hours"] for t in team)
|
|
total_fte = sum(t["fte"] for t in team)
|
|
adjusted_hours = sum(t["adjusted_hours"] for t in team)
|
|
adjusted_fte = sum(t["adjusted_fte"] for t in team)
|
|
programme_fte = sum(t["fte"] for t in team if t["is_programme_role"])
|
|
|
|
return {
|
|
"project_id": opp.id,
|
|
"project_name": opp.name,
|
|
"hours_per_fte": HOURS_PER_FTE,
|
|
"efficiency_pct": efficiency_pct,
|
|
"discipline_overrides": overrides,
|
|
"total_hours": round(total_hours, 2),
|
|
"total_fte": round(total_fte, 4),
|
|
"adjusted_hours": round(adjusted_hours, 2),
|
|
"adjusted_fte": round(adjusted_fte, 4),
|
|
"delivery_fte": round(total_fte - programme_fte, 4),
|
|
"programme_fte": round(programme_fte, 4),
|
|
"hours_saved": round(total_hours - adjusted_hours, 2),
|
|
"fte_saved": round(total_fte - adjusted_fte, 4),
|
|
"roles": team,
|
|
}
|