Wiki:Packs/Work Intake and Capacity/Modules
From WFM Labs
Modules for the Work Intake and Capacity pack (CP-OPS-007). Save the block under the filename in its heading and upload it as project knowledge with the rest of the pack. It needs only the Python standard library, plus openpyxl for Excel files.
Read Wiki:Packs/Work Intake and Capacity first.
Block 9 — intake.py
"""intake.py — Work Intake and Capacity: readout, capacity budget and flow metrics.
Reads the tracker export (CSV or XLSX, columns as in 06-tracker.md) and, optionally, last week's
export, the BAU register and the team roster. Every count in the readout is computed here, so the
narrative built on it never has to estimate one.
Usage:
python intake.py demo OUTDIR # write a synthetic team, register and 6 weeks of tracker
python intake.py readout THIS.csv [--prior LAST.csv] [--bau bau.csv] [--team team.csv]
[--asof YYYY-MM-DD] [--cap 5] [--out readout.md] [--xlsx readout.xlsx]
python intake.py capacity --bau bau.csv --team team.csv [--tracker THIS.csv]
python intake.py validate THIS.csv # codes, required fields, priority recompute
"""
from __future__ import annotations
import argparse
import csv
import datetime as dt
import random
import re
import statistics
from collections import Counter, defaultdict
from pathlib import Path
# ---------------------------------------------------------------- schema
CAPTURED = ["Date", "Channel", "Requester", "Requester role", "Work type", "Project", "Source", "Ask",
"Deliverable", "Due", "Urgency", "Alignment", "Size", "Clear?", "Open question"]
TRIAGE = ["Lane", "Owner", "Parent ID", "Priority", "Triage outcome"]
WORKING = ["Status", "Review date", "Est. hours", "Actual hours"]
CLOSE = ["Repeat of", "Outcome note", "Closed date"]
COLUMNS = ["ID"] + CAPTURED + TRIAGE + WORKING + CLOSE
CHOICES = {
"Channel": ["Email", "Zoom meeting", "Zoom chat", "In person", "Phone"],
"Work type": ["One-off", "Project"],
"Source": ["LL", "PO", "XF", "XO", "IN"],
"Urgency": ["U1", "U2", "U3", "U4"],
"Alignment": ["A1", "A2", "A3"],
"Size": ["XS", "S", "M", "L"],
"Clear?": ["Y", "N"],
"Triage outcome": ["Do now", "Assign", "Plan", "Park", "Redirect", "Decline"],
"Status": ["New", "Needs clarity", "Active", "On hold", "Backlog", "Program candidate", "Done", "Closed"],
}
OPEN = {"New", "Needs clarity", "Active", "On hold", "Backlog"}
ACTIVE = {"Active", "On hold"}
CLOSED = {"Done", "Closed"}
# Default hours per size band when no estimate or actual is logged. Mid-band, deliberately round.
SIZE_HOURS = {"XS": 0.5, "S": 4.0, "M": 20.0, "L": 60.0}
SYNONYMS = {"clear": "Clear?", "est hours": "Est. hours", "estimated hours": "Est. hours",
"actual hours": "Actual hours", "requester role": "Requester role", "work type": "Work type",
"triage outcome": "Triage outcome", "parent id": "Parent ID", "repeat of": "Repeat of",
"outcome note": "Outcome note", "closed date": "Closed date", "review date": "Review date",
"open question": "Open question", "created": "Date", "assigned to": "Owner"}
def priority(urgency: str, alignment: str) -> str:
"""The priority rule (04-classification.md). Mirrors the SharePoint calculated column."""
if urgency == "U1" or (alignment == "A1" and urgency == "U2"):
return "P1"
if alignment == "A1" or (alignment == "A2" and urgency == "U2"):
return "P2"
if alignment == "A2":
return "P3"
return "P4"
# ---------------------------------------------------------------- io
def _decode(h: str) -> str:
"""SharePoint internal names encode punctuation: Clear_x003f_ -> Clear?, Est_x002e__x0020_hours -> Est. hours."""
h = re.sub(r"_x([0-9a-fA-F]{4})_", lambda m: chr(int(m.group(1), 16)), h)
return re.sub(r"\s+([?.])", r"\1", re.sub(r"\s+", " ", h.strip()))
def _norm(h: str) -> str:
k = _decode(h)
low = k.lower().rstrip("?").rstrip(".")
for c in COLUMNS:
if c.lower().rstrip("?").rstrip(".") == low:
return c
return SYNONYMS.get(low, k)
def _heads(raw):
"""Canonical names. A synonym yields to a real column of that name ('Created' never overwrites
'Date'), and a later duplicate keeps its raw name."""
key = lambda h: _decode(h).lower().rstrip("?").rstrip(".")
canon = {key(c): c for c in COLUMNS}
present = {canon[key(h)] for h in raw if key(h) in canon}
out = []
for h in raw:
n = _norm(h)
if n in out or (n in present and key(h) not in canon):
n = h.strip()
out.append(n)
return out
def read_table(path: str) -> list[dict]:
p = Path(path)
if p.suffix.lower() in (".xlsx", ".xlsm"):
from openpyxl import load_workbook
ws = load_workbook(p, read_only=True, data_only=True).active
rows = list(ws.iter_rows(values_only=True))
head = _heads([str(h or "") for h in rows[0]])
out = [{h: ("" if v is None else v) for h, v in zip(head, r)} for r in rows[1:] if any(r)]
else:
with open(p, newline="", encoding="utf-8-sig") as f:
rd = csv.reader(f)
head = _heads(next(rd))
out = [dict(zip(head, r)) for r in rd if any(r)]
for r in out:
for k, v in list(r.items()):
if isinstance(v, (dt.date, dt.datetime)):
r[k] = (v.date() if isinstance(v, dt.datetime) else v).isoformat()
elif isinstance(v, float) and v.is_integer() and k == "ID":
r[k] = str(int(v))
else:
r[k] = str(v).strip()
return out
def write_csv(path, rows, cols):
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
w.writeheader()
w.writerows(rows)
DAYFIRST = False # set by --dayfirst: read 05/10/2026 as 5 October (UK/EU exports)
def d(s: str) -> dt.date | None:
"""Parse the date formats SharePoint and Excel exports produce; None if unreadable.
Slash dates are month-first unless DAYFIRST is set. validate() reports what fails."""
s = (s or "").strip()
if not s:
return None
if re.match(r"\d{4}-\d{2}-\d{2}", s):
try:
return dt.date.fromisoformat(s[:10])
except ValueError:
return None
md = ("%d/%m/%Y", "%m/%d/%Y") if DAYFIRST else ("%m/%d/%Y", "%d/%m/%Y")
fmts = [f + t for f in md for t in ("", " %H:%M", " %H:%M:%S", " %I:%M %p", " %I:%M:%S %p")]
fmts += ["%d %b %Y", "%d %B %Y", "%b %d, %Y", "%B %d, %Y", "%d-%b-%Y"]
for fmt in fmts:
try:
return dt.datetime.strptime(s, fmt).date()
except ValueError:
continue
return None
def hours(r: dict) -> float:
"""Actual if logged, else estimate, else the size band default."""
for k in ("Actual hours", "Est. hours"):
try:
v = float(r.get(k) or "nan")
if v == v:
return v
except ValueError:
pass
return SIZE_HOURS.get(r.get("Size", ""), 0.0)
def business_days(a: dt.date, b: dt.date) -> int:
"""Business days strictly after a up to and including b (0 if b <= a)."""
n, x = 0, a
while x < b:
x += dt.timedelta(days=1)
n += x.weekday() < 5
return n
# ---------------------------------------------------------------- validation
def validate(rows: list[dict]) -> list[str]:
issues = []
for r in rows:
rid = r.get("ID") or "?"
for col, allowed in CHOICES.items():
v = r.get(col, "")
if v and v not in allowed:
issues.append(f"ID {rid}: {col} = {v!r} not in {allowed}")
for col in ("Date", "Ask", "Work type", "Urgency", "Alignment", "Size", "Source"):
if not r.get(col):
issues.append(f"ID {rid}: {col} is empty")
if r.get("Urgency") and r.get("Alignment"):
want = priority(r["Urgency"], r["Alignment"])
if r.get("Priority") and r["Priority"] != want:
issues.append(f"ID {rid}: Priority {r['Priority']} but the rule gives {want}")
if r.get("Work type") == "Project" and not r.get("Project"):
issues.append(f"ID {rid}: Work type Project with no Project name")
if r.get("Clear?") == "Y" and not (r.get("Deliverable") and r.get("Due") and r.get("Requester")):
issues.append(f"ID {rid}: Clear? = Y but Deliverable, Due or Requester is missing")
if r.get("Status") in ACTIVE and not r.get("Owner"):
issues.append(f"ID {rid}: {r['Status']} with no Owner")
if r.get("Size") == "L" and r.get("Work type") == "One-off" and r.get("Status") not in ("Program candidate", *CLOSED):
issues.append(f"ID {rid}: L-size one-off; make it a Program candidate or re-scope")
if r.get("Status") in CLOSED and not r.get("Closed date"):
issues.append(f"ID {rid}: {r['Status']} with no Closed date")
for col in ("Date", "Due", "Review date", "Closed date"):
if r.get(col) and d(r[col]) is None:
issues.append(f"ID {rid}: {col} {r[col]!r} is not a readable date (try --dayfirst?)")
return issues
def fill_priority(rows):
for r in rows:
if r.get("Urgency") and r.get("Alignment"):
r["Priority"] = priority(r["Urgency"], r["Alignment"])
return rows
# ---------------------------------------------------------------- repeats
def repeat_chains(rows: list[dict]) -> dict[str, list[dict]]:
"""Group rows into chains by following 'Repeat of' to the root ID."""
by_id = {r.get("ID"): r for r in rows if r.get("ID")}
def root(r, seen=()):
ref = (r.get("Repeat of") or "").strip()
if ref and ref in by_id and ref not in seen:
return root(by_id[ref], seen + (ref,))
if ref and ref not in by_id and not seen:
return ref # the original is not in this export: the repeats still share it as a root
return r.get("ID") or ref
chains = defaultdict(list)
for r in rows:
if r.get("Repeat of"):
chains[root(r)].append(r)
for k in list(chains):
if k in by_id:
chains[k].insert(0, by_id[k])
return dict(chains)
def root_cause_triggers(rows, asof: dt.date, n=3, window_days=30) -> list[dict]:
"""Chains with n or more occurrences inside the window: each should open a root-cause one-off."""
out = []
for k, chain in repeat_chains(rows).items():
recent = [r for r in chain if (d(r.get("Date")) or asof) >= asof - dt.timedelta(days=window_days)]
if len(recent) >= n:
out.append(dict(root=k, count=len(recent), ask=chain[0].get("Ask", ""),
lanes=sorted({r.get("Lane", "") for r in chain} - {""}),
hours=round(sum(hours(r) for r in recent), 1)))
return sorted(out, key=lambda x: -x["count"])
# ---------------------------------------------------------------- flow (Little's Law)
def _plannable(r) -> bool:
return r.get("Size") != "L" and r.get("Urgency") != "U1"
def flow(rows, asof: dt.date, weeks: int = 4) -> dict:
"""Arrivals, completions, open work and lead time over the trailing window.
Little's Law: average items in the system = arrival rate x average time in the system.
Implied lead time = open items / completion rate; compare with the measured lead time of
items closed in the window. The measured figure counts only finished items, so it runs low
whenever work is accumulating. If arrivals exceed completions, the backlog grows by the gap
every week and every queued ask waits longer.
"""
start = asof - dt.timedelta(weeks=weeks)
arrived = [r for r in rows if (x := d(r.get("Date"))) and start < x <= asof]
# Every exit leaves the open pile: Done and Closed both count as completions.
done = [r for r in rows if r.get("Status") in CLOSED and (x := d(r.get("Closed date"))) and start < x <= asof]
open_ = [r for r in rows if r.get("Status") in OPEN]
lt = [business_days(d(r["Date"]), d(r["Closed date"])) for r in done if d(r.get("Date"))]
arr_w, done_w = len(arrived) / weeks, len(done) / weeks
return dict(
weeks=weeks, arrivals_per_week=round(arr_w, 1), completions_per_week=round(done_w, 1),
net_growth_per_week=round(arr_w - done_w, 1), open_items=len(open_),
implied_lead_time_weeks=round(len(open_) / done_w, 1) if done_w else None,
measured_lead_time_weeks_mean=round(statistics.mean(lt) / 5, 1) if lt else None,
measured_lead_time_bdays_median=statistics.median(lt) if lt else None,
# Hours on one basis for both sides: L-size asks become program candidates (not weekly
# demand), and U1 firefights are already netted out of capacity in the budget.
arrival_hours_per_week=round(sum(hours(r) for r in arrived if _plannable(r)) / weeks, 1),
completed_hours_per_week=round(sum(hours(r) for r in done if _plannable(r)) / weeks, 1),
repeat_share=round(sum(bool(r.get("Repeat of")) for r in arrived) / len(arrived), 2) if arrived else None,
)
# ---------------------------------------------------------------- capacity budget
def capacity(bau: list[dict], team: list[dict], rows: list[dict] | None = None,
asof: dt.date | None = None, weeks: int = 4) -> dict:
"""Weekly hours budget per person and lane.
team: Person, Lane, Weekly hours, Protected hours, Meeting hours (optional), Other hours (optional)
bau: Output, Lane, Owner, Cadence (daily/weekly/biweekly/monthly/quarterly), Hours per run
Discretionary = weekly hours - BAU - meetings - protected - other - measured U1 firefighting.
"""
per_week = {"daily": 5, "weekly": 1, "biweekly": 0.5, "fortnightly": 0.5, "monthly": 12 / 52,
"quarterly": 4 / 52, "annual": 1 / 52, "yearly": 1 / 52}
bau_by = Counter()
for b in bau:
mult = per_week.get((b.get("Cadence") or "").strip().lower())
if mult is None:
raise ValueError(f"BAU '{b.get('Output')}': unknown cadence {b.get('Cadence')!r}")
bau_by[b.get("Owner", "")] += float(b.get("Hours per run") or 0) * mult
fire = Counter()
if rows and asof:
start = asof - dt.timedelta(weeks=weeks)
for r in rows:
if r.get("Urgency") == "U1" and (x := d(r.get("Date"))) and start < x <= asof:
fire[r.get("Owner", "")] += hours(r) / weeks
people = []
for t in team:
name = t["Person"]
wk = float(t.get("Weekly hours") or 40)
parts = dict(bau=bau_by.get(name, 0.0), meetings=float(t.get("Meeting hours") or 0),
protected=float(t.get("Protected hours") or 0), other=float(t.get("Other hours") or 0),
firefighting=fire.get(name, 0.0))
people.append(dict(person=name, lane=t.get("Lane", ""), weekly=wk,
**{k: round(v, 1) for k, v in parts.items()},
discretionary=round(wk - sum(parts.values()), 1)))
roster = {t["Person"] for t in team}
unowned = round(sum(v for k, v in bau_by.items() if k not in roster), 1)
unowned_fire = round(sum(v for k, v in fire.items() if k not in roster), 1)
total = lambda k: round(sum(p[k] for p in people), 1)
return dict(people=people, unowned_bau=unowned, unowned_firefighting=unowned_fire,
totals={k: total(k) for k in ("weekly", "bau", "meetings", "protected", "other",
"firefighting", "discretionary")})
def supported_cap(discretionary_hours: float, target_weeks: float, typical_hours: float) -> float:
"""Active-item cap one person can carry and still finish a typical item in target_weeks.
A person splitting H discretionary hours a week across W active items gives each H/W hours a
week, so an item of h hours takes h*W/H weeks. Finishing within T weeks needs W <= H*T/h.
"""
if typical_hours <= 0:
return float("inf")
return discretionary_hours * target_weeks / typical_hours
# ---------------------------------------------------------------- readout
def _count(rows, col):
return Counter(r.get(col) or "(blank)" for r in rows)
def readout(rows, prior=None, bau=None, team=None, asof=None, cap=5) -> dict:
asof = asof or max((d(r.get("Date")) for r in rows if d(r.get("Date"))), default=dt.date.today())
fill_priority(rows)
week_start = asof - dt.timedelta(days=6)
open_ = [r for r in rows if r.get("Status") in OPEN]
active = [r for r in rows if r.get("Status") in ACTIVE]
new = [r for r in rows if (x := d(r.get("Date"))) and week_start <= x <= asof]
prior_ids = {r.get("ID") for r in (prior or [])}
prior_new = []
if prior:
p_asof = asof - dt.timedelta(days=7)
prior_new = [r for r in prior if (x := d(r.get("Date"))) and p_asof - dt.timedelta(days=6) <= x <= p_asof]
by_owner = Counter(r.get("Owner") or "(unassigned)" for r in active)
over = {k: v for k, v in by_owner.items() if v > cap}
overdue = [r for r in open_ if (x := d(r.get("Due"))) and x < asof]
clarifying = [r for r in rows if r.get("Status") == "Needs clarity" and (x := d(r.get("Date")))
and business_days(x, asof) > 1]
stale = [r for r in rows if r.get("Status") == "Backlog" and (x := d(r.get("Date")))
and (asof - x).days > 30]
wish = [r for r in rows if r.get("Status") == "Program candidate"
or (r.get("Size") == "L" and r.get("Status") in OPEN)]
def hours_by(col):
return _hours_by(rows, col, week_start, asof)
out = dict(
asof=asof.isoformat(), cap=cap,
snapshot=dict(open=len(open_), by_work_type=_count(open_, "Work type"), by_priority=_count(open_, "Priority"),
by_lane=_count(open_, "Lane"), by_status=_count(open_, "Status"),
by_project=_count([r for r in open_ if r.get("Work type") == "Project"], "Project"),
active_by_owner=dict(by_owner), over_cap=over),
intake=dict(new=len(new), prior_new=len(prior_new) if prior else None,
by_work_type=_count(new, "Work type"), by_source=_count(new, "Source"),
by_channel=_count(new, "Channel"), by_urgency=_count(new, "Urgency"),
unclear_share=round(sum(r.get("Clear?") == "N" for r in new) / len(new), 2) if new else None,
firefighting_share=round(sum(r.get("Urgency") == "U1" for r in new) / len(new), 2) if new else None,
carried_from_prior=len([r for r in open_ if r.get("ID") in prior_ids]) if prior else None),
aging=dict(overdue=[_brief(r) for r in overdue], clarifying_over_1bd=[_brief(r) for r in clarifying],
backlog_over_30d=[_brief(r) for r in stale]),
repeats=root_cause_triggers(rows, asof),
capacity=dict(hours_by_priority=hours_by("Priority"), hours_by_source=hours_by("Source"),
hours_by_lane=hours_by("Lane"),
logged_actual_share=round(sum(bool(r.get("Actual hours")) for r in rows if r.get("Status") == "Done")
/ max(1, sum(r.get("Status") == "Done" for r in rows)), 2)),
flow=flow(rows, asof),
candidates=[_brief(r) for r in wish],
)
if bau is not None and team is not None:
out["budget"] = capacity(bau, team, rows, asof)
return out
def _hours_by(rows, col, start, end):
c = Counter()
for r in rows:
x = d(r.get("Closed date")) if r.get("Status") == "Done" else None
if r.get("Status") in ACTIVE or (x and start <= x <= end):
c[r.get(col) or "(blank)"] += hours(r)
return {k: round(v, 1) for k, v in sorted(c.items())}
def _brief(r):
return {k: r.get(k, "") for k in ("ID", "Date", "Ask", "Owner", "Lane", "Priority", "Size", "Due", "Status", "Source")}
def to_markdown(ro: dict) -> str:
def cnt(c):
return ", ".join(f"{k} {v}" for k, v in sorted(c.items())) or "none"
def items(lst, n=12):
if not lst:
return "- none\n"
s = "".join(f"- #{r['ID']} {r['Ask'][:90]} ({r['Owner'] or 'no owner'}, {r['Priority']}, {r['Size']}"
f"{', due ' + r['Due'] if r['Due'] else ''})\n" for r in lst[:n])
return s + (f"- … and {len(lst) - n} more\n" if len(lst) > n else "")
s, i, a, f = ro["snapshot"], ro["intake"], ro["aging"], ro["flow"]
md = [f"# Weekly readout — week ending {ro['asof']}\n",
"Counts computed by intake.py from the tracker export. Hours use actuals where logged, else "
"estimates, else size-band defaults (XS 0.5, S 4, M 20, L 60).\n",
"## 1. Snapshot\n",
f"- Open items: **{s['open']}** — by work type: {cnt(s['by_work_type'])}; priority: {cnt(s['by_priority'])}",
f"- By lane: {cnt(s['by_lane'])}; by status: {cnt(s['by_status'])}",
f"- Projects: {cnt(s['by_project'])}",
f"- Active per person (cap {ro['cap']}): {cnt(s['active_by_owner'])}",
f"- **Over cap:** {cnt(s['over_cap'])}\n",
"## 2. This week's intake\n",
f"- New asks: **{i['new']}**" + (f" (last week {i['prior_new']})" if i["prior_new"] is not None else " (no prior export supplied)"),
f"- By work type: {cnt(i['by_work_type'])}; by source: {cnt(i['by_source'])}; by channel: {cnt(i['by_channel'])}",
f"- Urgency: {cnt(i['by_urgency'])}; firefighting (U1) share {i['firefighting_share']}; unclear at capture {i['unclear_share']}\n",
"## 3. Aging\n", "**Overdue**\n", items(a["overdue"]), "**In Needs clarity over 1 business day since asked**\n",
items(a["clarifying_over_1bd"]), "**Backlog over 30 days since asked**\n", items(a["backlog_over_30d"]),
"## 4. Repeats\n"]
md += [f"- Root #{t['root']}: {t['count']} occurrences in 30 days, {t['hours']} h — {t['ask'][:90]} "
"→ **open a root-cause one-off**" for t in ro["repeats"]] or ["- No chain reached 3 occurrences in 30 days"]
c = ro["capacity"]
md += ["\n## 5. Capacity\n",
f"- Hours (active + closed this week) by priority: {cnt(c['hours_by_priority'])}",
f"- By source: {cnt(c['hours_by_source'])}; by lane: {cnt(c['hours_by_lane'])}",
f"- Share of done items with actual hours logged: {c['logged_actual_share']}",
f"- Flow, trailing {f['weeks']} weeks: {f['arrivals_per_week']} arrivals/wk against "
f"{f['completions_per_week']} completions/wk (net {f['net_growth_per_week']:+}/wk); "
f"{f['open_items']} open; implied lead time {f['implied_lead_time_weeks']} weeks (Little's Law: open ÷ "
f"completions) against a measured mean of {f['measured_lead_time_weeks_mean']} weeks for items closed "
f"(median {f['measured_lead_time_bdays_median']} business days). The measured figure counts only "
"finished items, so it runs low whenever work is accumulating; read the two together with the "
"Backlog-over-30-days list",
f"- Repeat share: {f['repeat_share']} of asks in the window repeat an earlier one",
f"- Plannable demand {f['arrival_hours_per_week']} h/wk arriving (L-size and U1 excluded: L become "
"program candidates, U1 is netted from capacity below) "
f"against {f['completed_hours_per_week']} h/wk completed"]
if "budget" in ro:
t = ro["budget"]["totals"]
md.append(f"- Budget: {t['weekly']} h/wk total − BAU {t['bau']} − meetings {t['meetings']} − protected "
f"{t['protected']} − other {t['other']} − firefighting {t['firefighting']} = "
f"**{t['discretionary']} h/wk discretionary**")
dem = ro["flow"]["arrival_hours_per_week"]
if t["discretionary"] > 0:
md.append(f"- **Plannable demand is {dem} h/wk against {t['discretionary']} h/wk discretionary "
f"capacity ({dem / t['discretionary']:.0%}).** Above 100% the backlog must grow or something "
"must be deferred, declined or redirected.")
if ro["budget"]["unowned_firefighting"]:
md.append(f"- Firefighting by owners not on the roster: {ro['budget']['unowned_firefighting']} h/wk "
"(blank owner, or a name that does not match team.csv)")
if ro["budget"]["unowned_bau"]:
md.append(f"- BAU with no owner on the roster: {ro['budget']['unowned_bau']} h/wk")
md += ["\n## 6. Program candidates\n", items(ro["candidates"]),
"## 7. Decisions needed\n",
"_Written by the reviewer from sections 1–6; this script does not draft decisions._\n"]
return "\n".join(md)
def to_xlsx(ro: dict, rows: list[dict], path: str):
from openpyxl import Workbook
wb = Workbook()
ws = wb.active; ws.title = "tracker"
ws.append(COLUMNS)
for r in rows:
ws.append([r.get(c, "") for c in COLUMNS])
def sheet(name, recs):
w = wb.create_sheet(name)
if not recs:
w.append(["none"]); return
keys = list(recs[0].keys()); w.append(keys)
for x in recs:
w.append([str(x.get(k, "")) if isinstance(x.get(k), (list, dict)) else x.get(k, "") for k in keys])
sheet("overdue", ro["aging"]["overdue"])
sheet("backlog_30d", ro["aging"]["backlog_over_30d"])
sheet("repeats", ro["repeats"])
sheet("candidates", ro["candidates"])
sheet("flow", [ro["flow"]])
if "budget" in ro:
sheet("budget", ro["budget"]["people"])
wb.save(path)
# ---------------------------------------------------------------- synthetic demo
def demo(outdir: str, seed: int = 7, weeks: int = 6, asof: dt.date = dt.date(2026, 11, 13)):
"""Synthetic six-person team, three lanes, six weeks of asks. Not a benchmark."""
rng = random.Random(seed)
out = Path(outdir); out.mkdir(parents=True, exist_ok=True)
team = [dict(Person=p, Lane=l, **{"Weekly hours": 40, "Protected hours": 4, "Meeting hours": m, "Other hours": 2})
for p, l, m in [("Lead A", "Forecasting and capacity", 10), ("Analyst B", "Forecasting and capacity", 6),
("Lead C", "Scheduling and real time", 9), ("Analyst D", "Scheduling and real time", 5),
("Lead E", "Reporting and analytics", 9), ("Analyst F", "Reporting and analytics", 6)]]
bau = [dict(Output=o, Lane=l, Owner=w, Cadence=c, **{"Hours per run": h}) for o, l, w, c, h in [
("Daily intraday reforecast", "Scheduling and real time", "Analyst D", "daily", 2.0),
("Intraday staffing calls", "Scheduling and real time", "Lead C", "daily", 1.5),
("Weekly schedule publish", "Scheduling and real time", "Analyst D", "weekly", 5),
("Weekly volume forecast", "Forecasting and capacity", "Analyst B", "weekly", 6),
("Monthly capacity plan", "Forecasting and capacity", "Lead A", "monthly", 16),
("Daily performance report", "Reporting and analytics", "Analyst F", "daily", 1.5),
("Weekly business review pack", "Reporting and analytics", "Lead E", "weekly", 6),
("Monthly client scorecards", "Reporting and analytics", "Analyst F", "monthly", 12),
("Quarterly long-range plan", "Forecasting and capacity", "Lead A", "quarterly", 40)]]
lanes = {p["Person"]: p["Lane"] for p in team}
asks = ["Build a staffing scenario for a new client launch", "Explain last week's service-level miss",
"Pull handle-time trend by queue", "Re-run the forecast with a holiday shift",
"Size the overtime needed for the peak", "Add a chat queue to the daily report",
"Model the impact of a site move", "Answer a finance question on cost per contact",
"Reconcile two headcount numbers", "Review a vendor's capacity proposal",
"Draft a slide on absence trends", "Check whether a schedule change is feasible"]
projects = ["Platform migration", "New client onboarding"]
rows, nid, start = [], 1, asof - dt.timedelta(weeks=weeks)
recurring = {}
day = start
while day < asof:
day += dt.timedelta(days=1)
if day.weekday() >= 5:
continue
for _ in range(rng.choice([2, 3, 3, 4, 5])):
owner = rng.choice(list(lanes))
wt = rng.choices(["One-off", "Project"], [0.7, 0.3])[0]
u = rng.choices(["U1", "U2", "U3", "U4"], [0.12, 0.38, 0.25, 0.25])[0]
a = rng.choices(["A1", "A2", "A3"], [0.35, 0.45, 0.2])[0] if wt == "One-off" else "A1"
size = rng.choices(["XS", "S", "M", "L"], [0.55, 0.4, 0.05, 0.0] if u == "U1" else [0.3, 0.45, 0.2, 0.05])[0]
ask = rng.choice(asks)
rep = ""
if ask == "Reconcile two headcount numbers":
rep = recurring.setdefault(ask, str(nid)) if ask in recurring else ""
recurring.setdefault(ask, str(nid))
age = (asof - day).days
if u == "U1" or size == "XS":
status = "Done"
elif age > 21:
status = rng.choices(["Done", "Backlog", "Active", "Closed"], [0.6, 0.2, 0.1, 0.1])[0]
elif age > 7:
status = rng.choices(["Done", "Active", "On hold", "Backlog"], [0.35, 0.3, 0.1, 0.25])[0]
else:
status = rng.choices(["New", "Needs clarity", "Active", "Backlog"], [0.3, 0.2, 0.3, 0.2])[0]
if size == "L" and status not in CLOSED:
status = "Program candidate"
clear = rng.choices(["Y", "N"], [0.65, 0.35])[0]
if status == "Needs clarity":
clear = "N"
due = (day + dt.timedelta(days={"U1": 0, "U2": rng.randint(1, 6), "U3": rng.randint(8, 30)}.get(u, 0))) if u != "U4" else None
closed = None
if status in CLOSED:
lag = 0 if u == "U1" else rng.randint(1, {"XS": 2, "S": 8, "M": 18, "L": 35}[size])
closed = min(asof, day + dt.timedelta(days=lag))
r = dict(ID=str(nid), Date=day.isoformat(),
Channel=rng.choices(CHOICES["Channel"], [0.4, 0.3, 0.2, 0.08, 0.02])[0],
Requester=rng.choice(["Ops director", "Finance partner", "Client lead", "Line leader", "HR partner", "Own team"]),
**{"Requester role": "", "Work type": wt, "Project": rng.choice(projects) if wt == "Project" else "",
"Source": rng.choices(CHOICES["Source"], [0.15, 0.35, 0.2, 0.2, 0.1])[0], "Ask": ask,
"Deliverable": "file" if clear == "Y" else "", "Due": due.isoformat() if due and clear == "Y" else "",
"Urgency": u, "Alignment": a, "Size": size, "Clear?": clear,
"Open question": "" if clear == "Y" else "Deliverable and date", "Lane": lanes[owner],
"Owner": owner if status not in ("New",) else "", "Parent ID": "", "Priority": priority(u, a),
"Triage outcome": "Do now" if status == "Done" and u == "U1" else ("Plan" if status in ACTIVE else ""),
"Status": status, "Review date": "", "Est. hours": "",
"Actual hours": str(round(SIZE_HOURS[size] * rng.uniform(0.6, 1.6), 1)) if status == "Done" and size in ("M", "L") else "",
"Repeat of": rep, "Outcome note": "", "Closed date": closed.isoformat() if closed else ""})
if r["Clear?"] == "Y" and not r["Due"]:
r["Clear?"] = "N"; r["Open question"] = "Date"
rows.append(r); nid += 1
last_week = [r for r in rows if d(r["Date"]) <= asof - dt.timedelta(days=7)]
write_csv(out / "tracker-this-week.csv", rows, COLUMNS)
write_csv(out / "tracker-last-week.csv", last_week, COLUMNS)
write_csv(out / "team.csv", team, ["Person", "Lane", "Weekly hours", "Protected hours", "Meeting hours", "Other hours"])
write_csv(out / "bau-register.csv", bau, ["Output", "Lane", "Owner", "Cadence", "Hours per run"])
return dict(rows=len(rows), asof=asof.isoformat(), outdir=str(out))
# ---------------------------------------------------------------- cli
if __name__ == "__main__":
ap = argparse.ArgumentParser()
sub = ap.add_subparsers(dest="cmd", required=True)
s = sub.add_parser("demo"); s.add_argument("outdir")
s = sub.add_parser("validate"); s.add_argument("tracker")
s = sub.add_parser("readout")
s.add_argument("tracker"); s.add_argument("--prior"); s.add_argument("--bau"); s.add_argument("--team")
s.add_argument("--asof"); s.add_argument("--cap", type=int, default=5)
s.add_argument("--out"); s.add_argument("--xlsx")
s = sub.add_parser("capacity")
s.add_argument("--bau", required=True); s.add_argument("--team", required=True)
s.add_argument("--tracker"); s.add_argument("--asof"); s.add_argument("--target-weeks", type=float, default=1.0)
s.add_argument("--typical-hours", type=float, default=SIZE_HOURS["S"])
for sp in sub.choices.values():
sp.add_argument("--dayfirst", action="store_true", help="read slash dates as day/month/year")
a = ap.parse_args()
DAYFIRST = getattr(a, "dayfirst", False)
if a.cmd == "demo":
print(demo(a.outdir))
elif a.cmd == "validate":
iss = validate(read_table(a.tracker))
print("\n".join(iss) if iss else "no issues")
elif a.cmd == "readout":
rows = read_table(a.tracker)
ro = readout(rows, read_table(a.prior) if a.prior else None,
read_table(a.bau) if a.bau else None, read_table(a.team) if a.team else None,
d(a.asof) if a.asof else None, a.cap)
md = to_markdown(ro)
if a.out:
Path(a.out).write_text(md)
print(md)
if a.xlsx:
to_xlsx(ro, rows, a.xlsx)
elif a.cmd == "capacity":
rows = read_table(a.tracker) if a.tracker else None
asof = d(a.asof) if a.asof else (max(d(r["Date"]) for r in rows if d(r.get("Date"))) if rows else None)
b = capacity(read_table(a.bau), read_table(a.team), rows, asof)
print(f"{'person':12} {'lane':26} {'wk':>5} {'BAU':>5} {'mtg':>5} {'prot':>5} {'oth':>5} {'fire':>5} {'free':>6} {'cap':>5}")
for p in b["people"]:
capv = supported_cap(max(0, p["discretionary"]), a.target_weeks, a.typical_hours)
print(f"{p['person']:12} {p['lane'][:26]:26} {p['weekly']:5.0f} {p['bau']:5.1f} {p['meetings']:5.1f} "
f"{p['protected']:5.1f} {p['other']:5.1f} {p['firefighting']:5.1f} {p['discretionary']:6.1f} {capv:5.1f}")
t = b["totals"]
print(f"team: {t['weekly']} h/wk; BAU {t['bau']}; discretionary {t['discretionary']} "
f"({t['discretionary'] / t['weekly']:.0%}); unowned BAU {b['unowned_bau']} h/wk")
print(f"cap = active items a person can hold and still finish a {a.typical_hours:g} h item in "
f"{a.target_weeks:g} week(s)")
See also
- Wiki:Packs/Work Intake and Capacity — the pack this module belongs to
- Work Intake for Planning and Analytics Teams — the concepts
- Wiki:Packs — the pack index
