"""Lecture 15 teaching model: robot cell (cobot + CV/AI, on-prem, Jetson, ROS 2). Single source of truth for the Gantt, Monte Carlo, team cost and the Excel workbook. All durations in working days (wd); all money in EUR; all values are illustrative. Usage: python3 2026-11-03-robot-cell-model.py results.json (standard library only; ~10 s) Writes results.json (schedule, Monte Carlo P10–P90, costs) and mc_raw.json next to it.""" import json, math, random, statistics, sys, pathlib # ---- roles: (code, UK name, EN name, DE name, cost/day, sell/day) ------------- ROLES = [ ("PM", "Delivery / engineering manager", 300, 540), ("SA", "System architect", 340, 610), ("ROS", "Robotics engineer (ROS 2, MoveIt 2)", 300, 540), ("CV", "CV/ML engineer", 320, 580), ("SW", "Software engineer (HMI, on-prem, MLOps)", 260, 470), ("FW", "Firmware engineer", 280, 500), ("HW", "Hardware engineer", 280, 500), ("ME", "Mechanical engineer", 250, 450), ("SAF", "Safety engineer (part-time)", 360, 650), ("QA", "V&V / test engineer", 220, 400), ("TECH", "Technician (assembly, harness)", 140, 250), ("PROC", "Procurement manager", 180, 320), ] RATE = {r[0]: r for r in ROLES} # ---- WBS: id, epic, phase, name_en, O, M, P, deps, {role: FTE}, kind ----------- # kind: work | wait (calendar time with little effort) ; gate milestones attached by 'gate' T = [ # EPIC 1 — see & pick on the bench ("1.1", 1, "R&D", "Discovery: site, parts, IT policy", 5, 8, 15, [], {"SA":1, "CV":.5, "ROS":.5, "PM":.5}, "work", "G1"), ("1.2", 1, "R&D", "System requirements + acceptance", 5, 8, 12, ["1.1"], {"SA":1, "QA":.3, "SAF":.3, "PM":.3}, "work", "G2"), ("1.3", 1, "R&D", "Architecture (ROS graph, compute)", 5, 8, 12, ["1.2"], {"SA":1, "ROS":.5, "CV":.3, "SW":.3, "HW":.3}, "work", None), ("1.4", 1, "R&D", "HARA + project risk register", 2, 3, 5, ["1.3"], {"SAF":1, "SA":.5, "ROS":.3, "HW":.3, "ME":.3, "PM":.3}, "work", None), ("1.5", 1, "PoC", "PoC kit: Jetson, cameras, parts", 5, 10, 20, ["1.1"], {"PROC":.2}, "wait", None), ("1.6", 1, "PoC", "PoC-A: inference on Jetson", 10, 15, 25, ["1.4", "1.5"], {"CV":1, "SW":.3}, "work", None), ("1.7", 1, "PoC", "PoC-B: grasp with ROS 2 + MoveIt 2", 8, 12, 20, ["1.4", "1.5"], {"ROS":1, "ME":.3}, "work", None), ("1.8", 1, "PoC", "PoC review", 1, 2, 3, ["1.6", "1.7"], {"SA":.5, "PM":.5, "CV":.5, "ROS":.5}, "work", "G3"), ("1.9", 1, "MVP", "MVP: one part type, bench + sim", 10, 15, 22, ["1.8"], {"ROS":1, "CV":1, "SW":.5, "QA":.5}, "work", None), # EPIC 2 — tool & cell ("2.1", 2, "WP", "Detailed SW/HW/Mech requirements", 4, 6, 10, ["1.8"], {"SA":1, "HW":.5, "ME":.5, "FW":.3, "ROS":.3}, "work", None), ("2.2", 2, "WP", "HW load + component search", 3, 5, 10, ["2.1"], {"HW":1, "SA":.3, "PROC":.5}, "work", "G4"), ("2.3", 2, "WP", "Long-lead: cobot, Jetson, scanner", 20, 30, 50, ["2.2"], {"PROC":.2}, "wait", None), ("2.4", 2, "WP", "Tool PCB: schematic, PCB, BoM", 10, 15, 22, ["2.2"], {"HW":1}, "work", None), ("2.5", 2, "WP", "HW review (peer + FW)", 2, 3, 5, ["2.4"], {"HW":.5, "FW":.5}, "work", None), ("2.6", 2, "WP", "Mech design: tool, frame, guarding", 10, 15, 25, ["2.4"], {"ME":1, "HW":.2}, "work", None), ("2.7", 2, "WP", "Mech review + rework + all hands", 3, 5, 10, ["2.5", "2.6"], {"ME":.5, "HW":.5, "SA":.2}, "work", None), ("2.8", 2, "WP", "PCB manufacturing (vendor)", 15, 20, 30, ["2.7"], {"HW":.2}, "wait", None), ("2.9", 2, "WP", "Mech manufacturing (vendor)", 10, 18, 30, ["2.7"], {"ME":.2}, "wait", None), ("2.10", 2, "WP", "Standard parts + harness materials", 10, 15, 25, ["2.7"], {"PROC":.5}, "wait", None), ("2.11", 2, "WP", "Tool FW on dev board", 10, 15, 25, ["2.5"], {"FW":1}, "work", None), ("2.12", 2, "WP", "Cabinet + harness build", 8, 12, 20, ["2.10"], {"TECH":1, "HW":.5}, "work", None), ("2.13", 2, "WP", "Assembly + review + rework", 5, 8, 15, ["2.8", "2.9", "2.10"], {"TECH":1, "ME":.3, "HW":.3}, "work", None), ("2.14", 2, "WP", "FW on real HW + bench test", 4, 6, 10, ["2.11", "2.13"], {"FW":1, "QA":.5}, "work", None), ("2.15", 2, "WP", "Cell SW: orchestration, HMI, logs", 20, 30, 45, ["1.9"], {"ROS":1, "SW":1}, "work", None), ("2.16", 2, "WP", "Dataset v1 + 3 training iterations", 20, 30, 45, ["1.9"], {"CV":1, "SW":.3, "TECH":.2}, "work", None), ("2.17", 2, "WP", "On-prem GPU server, registry, OTA", 8, 12, 20, ["2.1"], {"SW":.5, "SA":.2}, "work", None), # EPIC 3 — integrate, validate, deliver ("3.1", 3, "WP", "Cell integration at our site", 10, 15, 25, ["2.3", "2.12", "2.14", "2.15", "2.17"], {"ROS":1, "CV":.5, "FW":.5, "HW":.5, "TECH":.5, "SW":.5}, "work", None), ("3.2", 3, "WP", "Safety validation", 5, 8, 15, ["3.1"], {"SAF":1, "ROS":.5, "QA":.5}, "work", None), ("3.3", 3, "WP", "Tuning: cycle time, accuracy", 8, 12, 20, ["3.1", "2.16"], {"CV":1, "ROS":1, "QA":.5}, "work", None), ("3.4", 3, "WP", "FAT with customer", 3, 4, 6, ["3.2", "3.3"], {"QA":1, "SA":.5, "PM":.5, "ROS":.5, "CV":.5}, "work", "FAT"), ("3.5", 3, "WP", "Shipping + customs", 5, 8, 15, ["3.4"], {"PROC":.2}, "wait", None), ("3.6", 3, "WP", "Installation + commissioning", 5, 8, 12, ["3.5"], {"ROS":1, "TECH":1, "CV":.5, "HW":.3}, "work", None), ("3.7", 3, "WP", "SAT + 2-week pilot run", 8, 10, 15, ["3.6"], {"QA":.5, "ROS":.5, "CV":.5, "PM":.3}, "work", "SAT"), ("3.8", 3, "Final", "Docs, training, handover", 4, 5, 8, ["3.7"], {"SA":.5, "QA":.5, "PM":.5, "SW":.3}, "work", None), ] PM_LOE = 0.5 # PM level-of-effort over the whole calendar, on top of task allocations # ---- project risks: id, task affected, probability, extra days, extra direct EUR RISKS = [ ("R1", "2.3", 0.30, 20, 0, "Long-lead slip: Jetson production module / cobot"), ("R2", "3.3", 0.40, 10, 0, "Accuracy below target on reflective parts: extra training iteration"), ("R3", "2.8", 0.30, 15, 1500, "Second PCB spin"), ("R4", "2.17", 0.40, 10, 0, "Customer IT security approval for on-prem server"), ("R5", "1.7", 0.25, 8, 0, "ROS 2 driver for the chosen cobot immature"), ("R6", "3.2", 0.25, 10, 3000, "Risk assessment demands extra safety device"), ("R7", "2.16", 0.10, 15, 0, "Key CV engineer leaves"), ("R8", "3.5", 0.40, 7, 500, "Shipping / customs delay"), ] # ---- non-labour (pass-through, T&M) -------------------------------------------- MATERIALS = [ # item, EUR, phase, group ("PoC kit: Jetson dev kit, 2 cameras, lighting, sample parts", 4600, "PoC", "Materials"), ("Collaborative 6-DoF robot (pilot cell)", 30000, "WP", "Materials"), ("Jetson production module + carrier + enclosure", 2500, "WP", "Materials"), ("Safety laser scanner + safety relay", 4500, "WP", "Materials"), ("Production cameras + lighting", 2800, "WP", "Materials"), ("On-prem GPU training server", 12000, "WP", "Materials"), ("Electrical cabinet components", 3000, "WP", "Materials"), ("End-effector BoM: vacuum, gripper actuators", 2500, "WP", "Materials"), ("PCB fabrication + assembly, 2 runs", 3000, "WP", "Vendors"), ("Mechanical fabrication: tool, frame, guarding", 9000, "WP", "Vendors"), ("Harness materials (built in-house)", 800, "WP", "Vendors"), ("Shipping, customs, logistics", 3500, "WP", "Vendors"), ] HANDLING_FEE = 0.08 # procurement handling fee on pass-through MGMT_RESERVE = 0.05 # management reserve on labour sell (unknown unknowns) FP_PREMIUM = 0.10 # premium for penalty exposure on FP phases PHASES = ["R&D", "PoC", "MVP", "WP", "Final"] MODEL = {"R&D": "FP", "PoC": "FP", "MVP": "T&M^", "WP": "T&M^ (engineering) + T&M (vendors)", "Final": "FP"} ids = [t[0] for t in T] TD = {t[0]: t for t in T} def pert(o, m, p): return (o + 4 * m + p) / 6, (p - o) / 6 def cpm(dur): es, ef = {}, {} for i in ids: # list is topologically ordered es[i] = max([ef[d] for d in TD[i][7]], default=0) ef[i] = es[i] + dur[i] finish = max(ef.values()) lf, ls = {}, {} succ = {i: [j for j in ids if i in TD[j][7]] for i in ids} for i in reversed(ids): lf[i] = min([ls[s] for s in succ[i]], default=finish) ls[i] = lf[i] - dur[i] fl = {i: round(ls[i] - es[i], 6) for i in ids} return es, ef, ls, lf, fl, finish def beta_pert(o, m, p, rng): if p == o: return m a = 1 + 4 * (m - o) / (p - o) b = 1 + 4 * (p - m) / (p - o) return o + rng.betavariate(a, b) * (p - o) def labour(dur, finish): """effort (person-days) by role and phase, given task durations and project finish.""" eff = {} for i in ids: ph = TD[i][2] for r, f in TD[i][8].items(): eff[(r, ph)] = eff.get((r, ph), 0) + f * dur[i] return eff def phase_spans(es, ef): sp = {} for i in ids: ph = TD[i][2] a, b = sp.get(ph, (1e9, -1)) sp[ph] = (min(a, es[i]), max(b, ef[i])) return sp def money(eff, finish, es, ef, kind="cost"): col = 2 if kind == "cost" else 3 by_phase = {ph: 0.0 for ph in PHASES} for (r, ph), d in eff.items(): by_phase[ph] += d * RATE[r][col] # PM LOE spread over phases in proportion to the calendar span they own sp = phase_spans(es, ef) total_span = sum(b - a for a, b in sp.values()) for ph, (a, b) in sp.items(): by_phase[ph] += PM_LOE * finish * (b - a) / total_span * RATE["PM"][col] return by_phase def deterministic(): E = {i: pert(*TD[i][4:7])[0] for i in ids} S = {i: pert(*TD[i][4:7])[1] for i in ids} M = {i: TD[i][5] for i in ids} es, ef, ls, lf, fl, fin = cpm(E) esm, efm, _, _, flm, finm = cpm(M) crit = [i for i in ids if abs(fl[i]) < 1e-6] eff = labour(E, fin) return dict(E=E, S=S, es=es, ef=ef, fl=fl, finish=fin, finish_M=finm, crit=crit, eff=eff, cost=money(eff, fin, es, ef, "cost"), sell=money(eff, fin, es, ef, "sell")) SYSTEMIC = (0.95, 1.0, 1.25) # shared optimism factor applied to all 'work' tasks in one run def monte_carlo(n=20000, seed=15, with_risks=True, systemic=True): rng = random.Random(seed) fins, costs, sells, risk_direct = [], [], [], [] ph_sell = {ph: [] for ph in PHASES} ph_end = {ph: [] for ph in PHASES} crit_count = {i: 0 for i in ids} for _ in range(n): dur = {i: beta_pert(*TD[i][4:7], rng) for i in ids} if systemic: k = rng.triangular(SYSTEMIC[0], SYSTEMIC[2], SYSTEMIC[1]) for i in ids: if TD[i][9] == "work": dur[i] *= k direct = 0 if with_risks: for rid, tid, p, days, eur, _ in RISKS: if rng.random() < p: dur[tid] += days direct += eur es, ef, ls, lf, fl, fin = cpm(dur) for i in ids: if abs(fl[i]) < 1e-6: crit_count[i] += 1 eff = labour(dur, fin) s = money(eff, fin, es, ef, "sell") c = money(eff, fin, es, ef, "cost") fins.append(fin); sells.append(sum(s.values())); costs.append(sum(c.values())); risk_direct.append(direct) sp = phase_spans(es, ef) for ph in PHASES: ph_sell[ph].append(s[ph]); ph_end[ph].append(sp[ph][1]) q = lambda xs, p: sorted(xs)[int(p * (len(xs) - 1))] return dict(fins=fins, P10=q(fins, .1), P50=q(fins, .5), P80=q(fins, .8), P90=q(fins, .9), mean=statistics.mean(fins), sell_P50=q(sells, .5), sell_P80=q(sells, .8), cost_P50=q(costs, .5), cost_P80=q(costs, .8), direct_mean=statistics.mean(risk_direct), ph_sell={ph: (q(v, .5), q(v, .8)) for ph, v in ph_sell.items()}, ph_end={ph: (q(v, .5), q(v, .8)) for ph, v in ph_end.items()}, crit_index={i: crit_count[i] / n for i in ids}) def weekly_fte(es, ef, dur): """FTE per role per calendar week (5 wd) using the deterministic schedule.""" weeks = int(math.ceil(max(ef.values()) / 5)) load = {r[0]: [0.0] * weeks for r in ROLES} for i in ids: for r, f in TD[i][8].items(): a, b = es[i], ef[i] for w in range(weeks): lo, hi = w * 5, (w + 1) * 5 ov = max(0, min(b, hi) - max(a, lo)) load[r][w] += f * ov / 5 for w in range(weeks): load["PM"][w] += PM_LOE return load if __name__ == "__main__": out = pathlib.Path(sys.argv[1]) d = deterministic() mc = monte_carlo() mc0 = monte_carlo(with_risks=False, systemic=False) mc1 = monte_carlo(with_risks=True, systemic=False) load = weekly_fte(d["es"], d["ef"], d["E"]) mat = {"Materials": sum(m[1] for m in MATERIALS if m[3] == "Materials"), "Vendors": sum(m[1] for m in MATERIALS if m[3] == "Vendors")} risk_emv_days = sum(p * days for _, _, p, days, _, _ in RISKS) risk_emv_eur = sum(p * eur for _, _, p, _, eur, _ in RISKS) eff_role = {} for (r, ph), v in d["eff"].items(): eff_role[r] = eff_role.get(r, 0) + v eff_role["PM"] += PM_LOE * d["finish"] res = dict( tasks=[dict(id=t[0], epic=t[1], phase=t[2], name=t[3], O=t[4], M=t[5], P=t[6], deps=t[7], roles=t[8], kind=t[9], gate=t[10], E=round(d["E"][t[0]], 2), S=round(d["S"][t[0]], 2), ES=round(d["es"][t[0]], 2), EF=round(d["ef"][t[0]], 2), float=round(d["fl"][t[0]], 2), crit_index=round(mc["crit_index"][t[0]], 3)) for t in T], finish_E=d["finish"], finish_M=d["finish_M"], critical=d["crit"], sum_M_all=sum(t[5] for t in T), mc={k: v for k, v in mc.items() if k not in ("fins",)}, mc_norisk={k: mc0[k] for k in ("P10", "P50", "P80", "P90", "mean")}, mc_indep={k: mc1[k] for k in ("P10", "P50", "P80", "P90", "mean", "sell_P50", "sell_P80")}, cost_det=d["cost"], sell_det=d["sell"], eff_role=eff_role, materials=mat, risk_emv_days=risk_emv_days, risk_emv_eur=risk_emv_eur, peak_fte=max(sum(load[r][w] for r in load) for w in range(len(load["PM"]))), weeks=len(load["PM"]), ) out.write_text(json.dumps(res, indent=1, ensure_ascii=False)) json.dump(dict(fins=mc["fins"], fins0=mc0["fins"], fins1=mc1["fins"], load=load), open(out.with_name("mc_raw.json"), "w")) print(json.dumps({k: res[k] for k in ("finish_E", "finish_M", "critical", "sum_M_all", "mc", "mc_norisk", "mc_indep", "cost_det", "sell_det", "eff_role", "materials", "risk_emv_days", "risk_emv_eur", "peak_fte", "weeks")}, indent=1))