mirror of
https://github.com/computerim/impactflow-discovery.git
synced 2026-08-27 06:00:35 +00:00
Joel Salmon
c4fc1cccd7
Build the boundary the ImpactFlow core time-tracker plugs into. A task maps to a foundation — one of the six stable profile elements (love, strength, mission, vocation, short_term, long_term) — so the tracker can ask "which goal does this build toward?" and post the answer back to Vision. - Models + migration 006: task_mapping (one row per logged time entry). - app/services/foundations.py: the six foundations + a pure, testable work-pattern aggregator (rollup) and a plain-language summary. - app/routers/integration.py (user-scoped; tracker calls as the user or via X-API-Key): GET /foundations, POST/GET /task-mappings, GET /work-patterns?days=N (per-foundation minutes/share/neglected). - Reminder engine now pulls from real work patterns: CheckinCoach takes an optional work-pattern summary (last 14 days) and reflects where time has gone against the person's own words — an observation, never a verdict. - Frontend: dashboard.html (time per foundation + neglected); linked from profile.html. Documented the core-tracker integration contract in the README. Phase 4 completes the Vision module's roadmap on the Discovery side; the core tracker integrates by calling these endpoints. Tests: 86 passing (added pure-aggregator, integration-endpoint, and work-pattern-into-check-in tests; run in-container). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
"""Phase 4 foundations: the six profile elements a task can build toward, and
|
||
a pure work-pattern aggregator.
|
||
|
||
A "foundation" is one of the stable parts of a person's profile. The core
|
||
time-tracker asks "which of these does this task build toward?" and posts the
|
||
mapping back; the aggregator rolls those mappings up per foundation to feed the
|
||
coaching reminder engine and the goal dashboard.
|
||
"""
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
# foundation key -> human label
|
||
FOUNDATIONS = {
|
||
"love": "What you love",
|
||
"strength": "What you're good at",
|
||
"mission": "What the world needs",
|
||
"vocation": "What you can be paid for",
|
||
"short_term": "Near-term goals (6–12mo)",
|
||
"long_term": "Long-term goals (3–5yr)",
|
||
}
|
||
|
||
# foundation key -> the DiscoveryProfile prose field it reflects
|
||
FOUNDATION_TO_FIELD = {
|
||
"love": "love_summary",
|
||
"strength": "strength_summary",
|
||
"mission": "mission_summary",
|
||
"vocation": "vocation_summary",
|
||
"short_term": "short_term_goals",
|
||
"long_term": "long_term_goals",
|
||
}
|
||
|
||
|
||
def rollup(entries: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""Aggregate task mappings into per-foundation work patterns.
|
||
|
||
Args:
|
||
entries: each ``{"foundation": str, "minutes": int,
|
||
"occurred_at": datetime}``. Already filtered to the desired window
|
||
by the caller.
|
||
|
||
Returns:
|
||
``{"total_minutes", "by_foundation": [...], "neglected": [...]}`` where
|
||
``by_foundation`` covers all six foundations (zero included), sorted by
|
||
minutes descending, and ``neglected`` lists foundations with no minutes.
|
||
"""
|
||
agg: Dict[str, Dict[str, Any]] = {
|
||
key: {"minutes": 0, "task_count": 0, "last_at": None}
|
||
for key in FOUNDATIONS
|
||
}
|
||
for e in entries:
|
||
key = e.get("foundation")
|
||
if key not in agg:
|
||
continue # ignore unknown foundations defensively
|
||
minutes = int(e.get("minutes") or 0)
|
||
agg[key]["minutes"] += minutes
|
||
agg[key]["task_count"] += 1
|
||
occurred = e.get("occurred_at")
|
||
if occurred is not None:
|
||
prev = agg[key]["last_at"]
|
||
if prev is None or occurred > prev:
|
||
agg[key]["last_at"] = occurred
|
||
|
||
total = sum(v["minutes"] for v in agg.values())
|
||
by_foundation = [
|
||
{
|
||
"foundation": key,
|
||
"label": FOUNDATIONS[key],
|
||
"minutes": v["minutes"],
|
||
"task_count": v["task_count"],
|
||
"last_at": v["last_at"],
|
||
"share": (v["minutes"] / total) if total else 0.0,
|
||
}
|
||
for key, v in agg.items()
|
||
]
|
||
by_foundation.sort(key=lambda r: r["minutes"], reverse=True)
|
||
neglected = [r["foundation"] for r in by_foundation if r["minutes"] == 0]
|
||
return {
|
||
"total_minutes": total,
|
||
"by_foundation": by_foundation,
|
||
"neglected": neglected,
|
||
}
|
||
|
||
|
||
def work_pattern_text(summary: Dict[str, Any], window_days: int) -> Optional[str]:
|
||
"""A short plain-language summary of recent work patterns for the coach to
|
||
reference. Returns None when there is no logged activity."""
|
||
if not summary or summary.get("total_minutes", 0) <= 0:
|
||
return None
|
||
active = [r for r in summary["by_foundation"] if r["minutes"] > 0]
|
||
spent = "; ".join(
|
||
f"{r['label']} {r['minutes']} min ({round(r['share'] * 100)}%)"
|
||
for r in active
|
||
)
|
||
parts = [f"In the last {window_days} days you logged time toward: {spent}."]
|
||
neglected = summary.get("neglected") or []
|
||
if neglected:
|
||
names = ", ".join(FOUNDATIONS[k] for k in neglected)
|
||
parts.append(f"Nothing was logged toward: {names}.")
|
||
return " ".join(parts)
|