scripts/time_block_planner.py
13.3 KB| 1 | #!/usr/bin/env python3 |
| 2 | """time_block_planner.py — Build a time-blocked day: deep work first, shallow batched, buffers everywhere. |
| 3 | |
| 4 | Encodes Cal Newport's time-block planning discipline as deterministic scheduling: |
| 5 | |
| 6 | 1. Deep tasks are scheduled FIRST, in the earliest hours of the day (energy-first). |
| 7 | Each deep task gets a block of at least 90 minutes — shorter blocks never reach depth. |
| 8 | 2. Total deep time is capped at 4 hours (240 min) — the trained practitioner's daily |
| 9 | ceiling. Demand beyond the cap is not squeezed in; the planner REFUSES (exit 2) |
| 10 | and names exactly which deep tasks to cut or defer. |
| 11 | 3. Shallow tasks are batched into at most two batches: one late morning (right after |
| 12 | the deep blocks) and one at the end of the day. A single shallow task goes to the |
| 13 | end-of-day batch so the morning stays protected. |
| 14 | 4. A 10-minute buffer separates consecutive work blocks (attention residue is real). |
| 15 | 5. An optional --lunch HH:MM inserts a fixed 30-minute break; blocks never straddle it. |
| 16 | 6. Shallow work that spills past --end is an overflow: the planner REFUSES (exit 2) |
| 17 | and names the overflow instead of silently extending the day. |
| 18 | |
| 19 | NO LLM CALLS. Pure arithmetic. Identical inputs always produce the identical plan. |
| 20 | |
| 21 | Exit codes: |
| 22 | 0 plan fits — markdown schedule (or --json) emitted |
| 23 | 1 usage / input error (bad time format, bad task spec, end <= start) |
| 24 | 2 refusal — deep demand exceeds the 4-hour cap, or the day overflows --end |
| 25 | |
| 26 | Usage: |
| 27 | python time_block_planner.py --start 08:30 --end 17:00 --lunch 12:30 \\ |
| 28 | --task "Write product spec:120:deep" --task "Email sweep:30:shallow" |
| 29 | python time_block_planner.py --sample |
| 30 | python time_block_planner.py --sample --json |
| 31 | """ |
| 32 | |
| 33 | import argparse |
| 34 | import json |
| 35 | import sys |
| 36 | from typing import Any, Dict, List, Optional |
| 37 | |
| 38 | DEEP_CAP_MIN = 240 # Newport's trained-limit ceiling: ~4 hours of deep work per day |
| 39 | MIN_DEEP_BLOCK = 90 # a deep block shorter than this never reaches depth |
| 40 | BUFFER_MIN = 10 # context-switch buffer between consecutive work blocks |
| 41 | LUNCH_MIN = 30 # fixed lunch break length |
| 42 | |
| 43 | |
| 44 | def parse_hhmm(value: str, flag: str) -> int: |
| 45 | parts = value.split(":") |
| 46 | if len(parts) != 2 or not parts[0].isdigit() or not parts[1].isdigit(): |
| 47 | raise ValueError(f"{flag} must be HH:MM, got: {value!r}") |
| 48 | h, m = int(parts[0]), int(parts[1]) |
| 49 | if h > 23 or m > 59: |
| 50 | raise ValueError(f"{flag} out of range: {value!r}") |
| 51 | return h * 60 + m |
| 52 | |
| 53 | |
| 54 | def fmt(minutes: int) -> str: |
| 55 | return f"{minutes // 60:02d}:{minutes % 60:02d}" |
| 56 | |
| 57 | |
| 58 | def fmt_dur(minutes: int) -> str: |
| 59 | h, m = divmod(minutes, 60) |
| 60 | if h and m: |
| 61 | return f"{h}h{m:02d}" |
| 62 | if h: |
| 63 | return f"{h}h" |
| 64 | return f"{m}min" |
| 65 | |
| 66 | |
| 67 | def parse_task(spec: str) -> Dict[str, Any]: |
| 68 | parts = spec.rsplit(":", 2) |
| 69 | if len(parts) != 3: |
| 70 | raise ValueError(f'task must be "name:minutes:deep|shallow", got: {spec!r}') |
| 71 | name, minutes_s, mode = parts[0].strip(), parts[1].strip(), parts[2].strip().lower() |
| 72 | if not name: |
| 73 | raise ValueError(f"task name is empty in: {spec!r}") |
| 74 | if not minutes_s.isdigit() or int(minutes_s) <= 0: |
| 75 | raise ValueError(f"task minutes must be a positive integer in: {spec!r}") |
| 76 | if mode not in ("deep", "shallow"): |
| 77 | raise ValueError(f'task mode must be "deep" or "shallow" in: {spec!r}') |
| 78 | return {"name": name, "minutes": int(minutes_s), "mode": mode} |
| 79 | |
| 80 | |
| 81 | def _ev(start: int, end: int, label: str, kind: str) -> Dict[str, Any]: |
| 82 | return {"start": start, "end": end, "label": label, "kind": kind, |
| 83 | "minutes": end - start} |
| 84 | |
| 85 | |
| 86 | def plan(start: int, end: int, tasks: List[Dict[str, Any]], |
| 87 | lunch: Optional[int]) -> Dict[str, Any]: |
| 88 | """Return a plan dict. verdict is PLANNED, DEEP-CAP-EXCEEDED, or OVERFLOW.""" |
| 89 | deep = [t for t in tasks if t["mode"] == "deep"] |
| 90 | shallow = [t for t in tasks if t["mode"] == "shallow"] |
| 91 | notes: List[str] = [] |
| 92 | |
| 93 | # --- 1. Deep blocks: at least 90 min each, earliest hours, capped at 4h total --- |
| 94 | deep_blocks = [] |
| 95 | for t in deep: |
| 96 | dur = max(MIN_DEEP_BLOCK, t["minutes"]) |
| 97 | if dur > t["minutes"]: |
| 98 | notes.append(f'"{t["name"]}" widened from {t["minutes"]} to {dur} min ' |
| 99 | f"(deep blocks below {MIN_DEEP_BLOCK} min never reach depth).") |
| 100 | deep_blocks.append({"name": t["name"], "dur": dur}) |
| 101 | total_deep = sum(b["dur"] for b in deep_blocks) |
| 102 | if total_deep > DEEP_CAP_MIN: |
| 103 | acc, cut = 0, [] |
| 104 | for b in deep_blocks: |
| 105 | if acc + b["dur"] <= DEEP_CAP_MIN: |
| 106 | acc += b["dur"] |
| 107 | else: |
| 108 | cut.append(b["name"]) |
| 109 | return { |
| 110 | "verdict": "DEEP-CAP-EXCEEDED", |
| 111 | "reason": (f"Deep demand is {fmt_dur(total_deep)} but the trained daily ceiling " |
| 112 | f"is {fmt_dur(DEEP_CAP_MIN)}. Depth past the cap is fake depth — " |
| 113 | f"cut or defer to another day: " + ", ".join(f'"{n}"' for n in cut)), |
| 114 | "deep_demand_minutes": total_deep, |
| 115 | "deep_cap_minutes": DEEP_CAP_MIN, |
| 116 | "cut_or_defer": cut, |
| 117 | } |
| 118 | |
| 119 | # --- 2. Shallow batches: at most two (late morning + end of day) --- |
| 120 | if len(shallow) >= 2: |
| 121 | n_a = (len(shallow) + 1) // 2 |
| 122 | batch_a, batch_b = shallow[:n_a], shallow[n_a:] |
| 123 | elif len(shallow) == 1: |
| 124 | batch_a, batch_b = [], shallow[:] # single batch goes end-of-day |
| 125 | else: |
| 126 | batch_a, batch_b = [], [] |
| 127 | dur_a = sum(t["minutes"] for t in batch_a) |
| 128 | dur_b = sum(t["minutes"] for t in batch_b) |
| 129 | |
| 130 | # --- 3. Forward pass: deep blocks, then shallow batch A, lunch-aware --- |
| 131 | events: List[Dict[str, Any]] = [] |
| 132 | ctx = {"lunch": lunch, "lunch_placed": lunch is None} |
| 133 | cursor = start |
| 134 | |
| 135 | def advance(cursor: int, dur: int, label: str, kind: str) -> int: |
| 136 | if events and events[-1]["kind"] in ("deep", "shallow"): |
| 137 | events.append(_ev(cursor, cursor + BUFFER_MIN, "Buffer — stand up, reset", "buffer")) |
| 138 | cursor += BUFFER_MIN |
| 139 | if not ctx["lunch_placed"]: |
| 140 | if cursor >= ctx["lunch"]: |
| 141 | events.append(_ev(cursor, cursor + LUNCH_MIN, "Lunch — away from the desk", "break")) |
| 142 | cursor += LUNCH_MIN |
| 143 | ctx["lunch_placed"] = True |
| 144 | elif cursor + dur > ctx["lunch"]: |
| 145 | if ctx["lunch"] > cursor: |
| 146 | events.append(_ev(cursor, ctx["lunch"], "Flex — reset, no inputs", "flex")) |
| 147 | events.append(_ev(ctx["lunch"], ctx["lunch"] + LUNCH_MIN, |
| 148 | "Lunch — away from the desk", "break")) |
| 149 | cursor = ctx["lunch"] + LUNCH_MIN |
| 150 | ctx["lunch_placed"] = True |
| 151 | events.append(_ev(cursor, cursor + dur, label, kind)) |
| 152 | return cursor + dur |
| 153 | |
| 154 | for b in deep_blocks: |
| 155 | cursor = advance(cursor, b["dur"], f'DEEP — {b["name"]}', "deep") |
| 156 | if batch_a: |
| 157 | label_a = "SHALLOW batch (late morning) — " + " · ".join(t["name"] for t in batch_a) |
| 158 | cursor = advance(cursor, dur_a, label_a, "shallow") |
| 159 | |
| 160 | # --- 4. Backward pass: shallow batch B ends exactly at --end --- |
| 161 | # The 10-min buffer invariant applies before batch B too: when batch B would |
| 162 | # directly follow a work block, that buffer is part of the day's budget. |
| 163 | need_buffer = (BUFFER_MIN if batch_b and events and events[-1]["kind"] in ("deep", "shallow") |
| 164 | else 0) |
| 165 | if cursor + need_buffer + dur_b > end: |
| 166 | overflow = cursor + need_buffer + dur_b - end |
| 167 | return { |
| 168 | "verdict": "OVERFLOW", |
| 169 | "reason": (f"The day overflows {fmt(end)} by {overflow} min. " |
| 170 | + (("Defer shallow work instead of extending the day: " |
| 171 | + ", ".join(f'"{t["name"]}"' for t in (batch_b or batch_a))) |
| 172 | if (batch_b or batch_a) else |
| 173 | "Deep demand alone exceeds the day — trim a deep block or extend --end.")), |
| 174 | "overflow_minutes": overflow, |
| 175 | "defer_candidates": [t["name"] for t in (batch_b or batch_a)], |
| 176 | } |
| 177 | if need_buffer: |
| 178 | events.append(_ev(cursor, cursor + BUFFER_MIN, "Buffer — stand up, reset", "buffer")) |
| 179 | cursor += BUFFER_MIN |
| 180 | start_b = end - dur_b |
| 181 | if start_b > cursor: |
| 182 | if not ctx["lunch_placed"] and cursor <= ctx["lunch"] and ctx["lunch"] + LUNCH_MIN <= start_b: |
| 183 | if ctx["lunch"] > cursor: |
| 184 | events.append(_ev(cursor, ctx["lunch"], "Flex — overflow absorber", "flex")) |
| 185 | events.append(_ev(ctx["lunch"], ctx["lunch"] + LUNCH_MIN, |
| 186 | "Lunch — away from the desk", "break")) |
| 187 | cursor = ctx["lunch"] + LUNCH_MIN |
| 188 | ctx["lunch_placed"] = True |
| 189 | if start_b > cursor: |
| 190 | events.append(_ev(cursor, start_b, "Flex — overflow absorber", "flex")) |
| 191 | if not ctx["lunch_placed"]: |
| 192 | notes.append(f"Lunch could not be placed at {fmt(lunch)} without splitting a block — " |
| 193 | "move it or shorten a block.") |
| 194 | if batch_b: |
| 195 | label_b = "SHALLOW batch (end of day) — " + " · ".join(t["name"] for t in batch_b) |
| 196 | events.append(_ev(start_b, end, label_b, "shallow")) |
| 197 | |
| 198 | totals = {} |
| 199 | for e in events: |
| 200 | totals[e["kind"]] = totals.get(e["kind"], 0) + e["minutes"] |
| 201 | return { |
| 202 | "verdict": "PLANNED", |
| 203 | "start": fmt(start), |
| 204 | "end": fmt(end), |
| 205 | "events": [{"start": fmt(e["start"]), "end": fmt(e["end"]), |
| 206 | "label": e["label"], "kind": e["kind"], "minutes": e["minutes"]} |
| 207 | for e in events], |
| 208 | "totals_minutes": totals, |
| 209 | "deep_total_minutes": totals.get("deep", 0), |
| 210 | "deep_cap_minutes": DEEP_CAP_MIN, |
| 211 | "notes": notes, |
| 212 | } |
| 213 | |
| 214 | |
| 215 | def render_human(r: Dict[str, Any]) -> str: |
| 216 | out: List[str] = [] |
| 217 | if r["verdict"] != "PLANNED": |
| 218 | out.append(f"REFUSED — {r['verdict']}") |
| 219 | out.append("=" * 64) |
| 220 | out.append(f" {r['reason']}") |
| 221 | return "\n".join(out) |
| 222 | out.append(f"## Time-Block Plan — {r['start']} → {r['end']}") |
| 223 | out.append("") |
| 224 | out.append("| Start | End | Block | Mode |") |
| 225 | out.append("|-------|-----|-------|------|") |
| 226 | for e in r["events"]: |
| 227 | out.append(f"| {e['start']} | {e['end']} | {e['label']} | {e['kind'].upper()} |") |
| 228 | out.append("") |
| 229 | t = r["totals_minutes"] |
| 230 | out.append(f"Deep {fmt_dur(t.get('deep', 0))} / {fmt_dur(r['deep_cap_minutes'])} cap · " |
| 231 | f"Shallow {fmt_dur(t.get('shallow', 0))} · " |
| 232 | f"Flex {fmt_dur(t.get('flex', 0))} · " |
| 233 | f"Buffers {fmt_dur(t.get('buffer', 0))}") |
| 234 | for n in r["notes"]: |
| 235 | out.append(f"- Note: {n}") |
| 236 | out.append("") |
| 237 | out.append("Flex absorbs what the plan didn't foresee — when the day breaks, " |
| 238 | "revise the blocks; never abandon them.") |
| 239 | return "\n".join(out) |
| 240 | |
| 241 | |
| 242 | SAMPLE = { |
| 243 | "start": "08:30", "end": "17:00", "lunch": "12:30", |
| 244 | "tasks": [ |
| 245 | "Write product spec:120:deep", |
| 246 | "Design onboarding flow:90:deep", |
| 247 | "Email sweep:30:shallow", |
| 248 | "Team status update:20:shallow", |
| 249 | "Expense report:15:shallow", |
| 250 | ], |
| 251 | } |
| 252 | |
| 253 | |
| 254 | def main(argv: List[str]) -> int: |
| 255 | p = argparse.ArgumentParser( |
| 256 | description="Build a time-blocked day: deep work first, shallow batched, buffers everywhere.", |
| 257 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 258 | epilog=("Exit codes:\n" |
| 259 | " 0 plan fits — schedule emitted\n" |
| 260 | " 1 usage / input error\n" |
| 261 | " 2 refusal — deep demand exceeds the 4-hour cap, or the day overflows --end"), |
| 262 | ) |
| 263 | p.add_argument("--start", help="Day start, HH:MM") |
| 264 | p.add_argument("--end", help="Day end, HH:MM (hard stop — fixed-schedule productivity)") |
| 265 | p.add_argument("--task", action="append", default=[], |
| 266 | help='Repeatable: "name:minutes:deep|shallow"') |
| 267 | p.add_argument("--lunch", help="Optional lunch start, HH:MM (fixed 30-min break)") |
| 268 | p.add_argument("--json", action="store_true", help="Emit JSON instead of markdown") |
| 269 | p.add_argument("--sample", action="store_true", help="Run the embedded sample day") |
| 270 | args = p.parse_args(argv) |
| 271 | |
| 272 | if args.sample: |
| 273 | args.start, args.end = SAMPLE["start"], SAMPLE["end"] |
| 274 | args.lunch, args.task = SAMPLE["lunch"], SAMPLE["tasks"] |
| 275 | if not args.start or not args.end or not args.task: |
| 276 | p.print_help() |
| 277 | print("\nerror: --start, --end and at least one --task are required (or --sample)", |
| 278 | file=sys.stderr) |
| 279 | return 1 |
| 280 | try: |
| 281 | start = parse_hhmm(args.start, "--start") |
| 282 | end = parse_hhmm(args.end, "--end") |
| 283 | lunch = parse_hhmm(args.lunch, "--lunch") if args.lunch else None |
| 284 | if end <= start: |
| 285 | raise ValueError("--end must be after --start") |
| 286 | tasks = [parse_task(s) for s in args.task] |
| 287 | except ValueError as exc: |
| 288 | print(f"error: {exc}", file=sys.stderr) |
| 289 | return 1 |
| 290 | |
| 291 | result = plan(start, end, tasks, lunch) |
| 292 | if args.json: |
| 293 | print(json.dumps(result, indent=2)) |
| 294 | else: |
| 295 | print(render_human(result)) |
| 296 | return 0 if result["verdict"] == "PLANNED" else 2 |
| 297 | |
| 298 | |
| 299 | if __name__ == "__main__": |
| 300 | sys.exit(main(sys.argv[1:])) |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
