"""Roadmap graph read models — AP1.4d/4e.""" from __future__ import annotations from typing import Any, Optional from steering.graph.profiles import ( DEFAULT_METHOD_KEY, GraphMethodProfile, get_graph_profile, graph_profile_as_dict, ) from steering.graph.join_branch import apply_deferred_topology TERMINAL_STATUSES = frozenset({"reached", "moved", "discarded"}) ACTIVE_STATUSES = frozenset({"planned", "active", "at_risk"}) BLOCKING_STATUSES = frozenset({"planned", "active", "at_risk"}) def _is_prerequisite_satisfied(status: str) -> bool: return status == "reached" def _fulfillment_ratio(progress: Optional[dict[str, int]]) -> Optional[float]: if not progress: return None total = progress.get("total", 0) if total <= 0: return None closed = progress.get("closed", 0) return round(closed / total, 4) def _sort_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]: return sorted( items, key=lambda item: ( item.get("sort_order", 0), item.get("title") or "", str(item.get("id", "")), ), ) def _sequential_order_blocked_by( item: dict[str, Any], sorted_items: list[dict[str, Any]], ) -> list[str]: """Fallback wenn keine Kanten: sequenzielle Items nach sort_order.""" blocked_by: list[str] = [] item_id = str(item["id"]) for prev in sorted_items: if str(prev["id"]) == item_id: break if prev.get("sequencing_mode", "sequential") != "sequential": continue if not _is_prerequisite_satisfied(prev.get("status", "planned")): blocked_by.append(str(prev["id"])) return blocked_by def compute_initiative_graph_state( *, items: list[dict[str, Any]], dependencies: list[dict[str, Any]], criteria_progress: Optional[dict[str, dict[str, int]]] = None, ) -> dict[str, Any]: """ Read Model pro Initiative: blocked / ready / fulfillment_ratio. Kanten-Semantik (from → to): - requires: from benötigt to (Voraussetzung) als reached - optional_branch: wie requires, blockiert aber nicht (nur Metadaten) - blocks: from blockiert to solange from noch aktiv ist - parallel_group / related: keine Blockade """ criteria_progress = criteria_progress or {} sorted_items = _sort_items(items) item_by_id = {str(item["id"]): item for item in items} has_any_deps = len(dependencies) > 0 blocked_by_map: dict[str, list[str]] = {item_id: [] for item_id in item_by_id} for dep in dependencies: dep_type = dep.get("dependency_type") or dep.get("edge_kind") or "requires" from_id = str(dep["from_item_id"]) to_id = str(dep["to_item_id"]) if from_id not in item_by_id or to_id not in item_by_id: continue if dep_type == "requires": prereq = item_by_id[to_id] if not _is_prerequisite_satisfied(prereq.get("status", "planned")): if to_id not in blocked_by_map[from_id]: blocked_by_map[from_id].append(to_id) elif dep_type == "blocks": blocker = item_by_id[from_id] if blocker.get("status", "planned") in BLOCKING_STATUSES: if from_id not in blocked_by_map[to_id]: blocked_by_map[to_id].append(from_id) if not has_any_deps: for item in sorted_items: item_id = str(item["id"]) if item.get("sequencing_mode", "sequential") == "sequential": for prereq_id in _sequential_order_blocked_by(item, sorted_items): if prereq_id not in blocked_by_map[item_id]: blocked_by_map[item_id].append(prereq_id) blocked_by_map = apply_deferred_topology( dependencies=dependencies, item_by_id=item_by_id, blocked_by_map=blocked_by_map, ) item_states: dict[str, dict[str, Any]] = {} blocked_items: list[str] = [] ready_items: list[str] = [] for item_id, item in item_by_id.items(): status = item.get("status", "planned") blocked_by = blocked_by_map.get(item_id, []) blocked = len(blocked_by) > 0 ready = ( status in ACTIVE_STATUSES and not blocked ) ratio = _fulfillment_ratio(criteria_progress.get(item_id)) item_states[item_id] = { "status": status, "blocked": blocked, "ready": ready, "blocked_by": blocked_by, "fulfillment_ratio": ratio, } if blocked and status in ACTIVE_STATUSES: blocked_items.append(item_id) if ready: ready_items.append(item_id) return { "items": item_states, "blocked_items": blocked_items, "ready_items": ready_items, } def _initiative_fulfillment_ratio(item_states: dict[str, dict[str, Any]]) -> Optional[float]: ratios = [ st["fulfillment_ratio"] for st in item_states.values() if st.get("fulfillment_ratio") is not None ] if not ratios: return None return round(sum(ratios) / len(ratios), 4) def apply_graph_method_profile( state: dict[str, Any], profile: GraphMethodProfile ) -> dict[str, Any]: """Passt blocked/ready an Methodenprofil an (AP1.4e).""" if profile.enforce_gate_blocking: state["graph_profile"] = graph_profile_as_dict(profile) state["initiative_fulfillment_ratio"] = _initiative_fulfillment_ratio(state["items"]) return state item_states = state["items"] blocked_items: list[str] = [] ready_items: list[str] = [] for item_id, item_state in item_states.items(): would_block = item_state["blocked"] item_state["would_block"] = would_block item_state["blocked"] = False status = item_state.get("status", "planned") item_state["ready"] = status in ACTIVE_STATUSES if item_state["ready"]: ready_items.append(item_id) state["items"] = item_states state["blocked_items"] = blocked_items state["ready_items"] = ready_items state["graph_profile"] = graph_profile_as_dict(profile) state["initiative_fulfillment_ratio"] = _initiative_fulfillment_ratio(item_states) return state def compute_initiative_graph_state_with_profile( *, items: list[dict[str, Any]], dependencies: list[dict[str, Any]], criteria_progress: Optional[dict[str, dict[str, int]]] = None, method_key: Optional[str] = None, ) -> dict[str, Any]: base = compute_initiative_graph_state( items=items, dependencies=dependencies, criteria_progress=criteria_progress, ) profile = get_graph_profile(method_key) result = apply_graph_method_profile(base, profile) result["method_key"] = method_key or DEFAULT_METHOD_KEY return result def load_initiative_graph_state( *, tenant_id: str, initiative_id: str ) -> dict[str, Any]: """Lädt Items, Kanten, Kriterien und Methodenprofil.""" from services import roadmap as roadmap_service from services import roadmap_criteria as criteria_service from services.steering_context import get_steering_context items = roadmap_service.list_roadmap_items_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) dependencies = roadmap_service.list_dependencies_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) criteria_progress = criteria_service.criteria_progress_for_initiative( tenant_id=tenant_id, initiative_id=initiative_id ) steering = get_steering_context(tenant_id=tenant_id, initiative_id=initiative_id) method_key = (steering or {}).get("method_key") or DEFAULT_METHOD_KEY return compute_initiative_graph_state_with_profile( items=items, dependencies=dependencies, criteria_progress=criteria_progress, method_key=method_key, )