Spaces:
Paused
Paused
File size: 4,507 Bytes
8697ae6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """HCM:21 phase management per quarter.
Each quarter cycles through four phases:
Scanning → Planning → Producing → Controlling
Each phase has minimum action requirements before the next phase unlocks.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Set
PHASES = ["scanning", "planning", "producing", "controlling"]
PHASE_ACTIONS: Dict[str, List[str]] = {
"scanning": [
"query_department",
"query_employees",
"calculate_metric",
"review_financials",
],
"planning": [
"set_hiring_target",
"set_training_budget",
"set_compensation_policy",
"set_retention_program",
],
"producing": [
"execute_hiring",
"execute_promotion",
"execute_transfer",
"execute_training",
"execute_termination",
],
"controlling": [
"submit_report",
"calculate_metric",
],
}
# Minimum actions required in a phase before advancing
PHASE_MIN_ACTIONS = {
"scanning": 2,
"planning": 1,
"producing": 1,
"controlling": 1,
}
# Control actions available in any phase
ALWAYS_AVAILABLE = ["advance_phase", "advance_quarter"]
@dataclass
class PhaseManager:
"""Tracks phase state within a quarter."""
current_phase: str = "scanning"
actions_in_phase: int = 0
phase_history: Dict[str, int] = field(default_factory=lambda: {p: 0 for p in PHASES})
phases_completed: Set[str] = field(default_factory=set)
def get_available_actions(self) -> List[str]:
"""Get list of valid action types for the current phase."""
actions = list(PHASE_ACTIONS.get(self.current_phase, []))
# advance_phase available if minimum actions met
if self.can_advance_phase():
actions.append("advance_phase")
# advance_quarter available if we're in controlling phase with min actions met
if self.current_phase == "controlling" and self.actions_in_phase >= PHASE_MIN_ACTIONS["controlling"]:
actions.append("advance_quarter")
# submit_report only in controlling
return actions
def is_valid_action(self, action_type: str) -> bool:
"""Check if an action type is valid in the current phase."""
return action_type in self.get_available_actions()
def can_advance_phase(self) -> bool:
"""Check if the minimum action requirement is met for advancing."""
min_required = PHASE_MIN_ACTIONS.get(self.current_phase, 1)
return self.actions_in_phase >= min_required
def advance_phase(self) -> str:
"""Move to the next phase. Returns new phase name.
Raises ValueError if minimum actions not met or already in last phase.
"""
if not self.can_advance_phase():
min_req = PHASE_MIN_ACTIONS.get(self.current_phase, 1)
raise ValueError(
f"Cannot advance: need {min_req} actions in {self.current_phase} phase, "
f"only {self.actions_in_phase} taken."
)
current_idx = PHASES.index(self.current_phase)
if current_idx >= len(PHASES) - 1:
raise ValueError("Already in the final phase (controlling). Use advance_quarter instead.")
self.phases_completed.add(self.current_phase)
self.current_phase = PHASES[current_idx + 1]
self.actions_in_phase = 0
return self.current_phase
def record_action(self, action_type: str) -> None:
"""Record that an action was taken in the current phase."""
if action_type not in ("advance_phase", "advance_quarter"):
self.actions_in_phase += 1
self.phase_history[self.current_phase] = self.phase_history.get(self.current_phase, 0) + 1
def reset_for_new_quarter(self) -> None:
"""Reset phase state for a new quarter."""
self.current_phase = "scanning"
self.actions_in_phase = 0
self.phase_history = {p: 0 for p in PHASES}
self.phases_completed = set()
def get_phase_summary(self) -> dict:
"""Get summary of phase progress."""
return {
"current_phase": self.current_phase,
"actions_in_phase": self.actions_in_phase,
"min_required": PHASE_MIN_ACTIONS.get(self.current_phase, 1),
"can_advance": self.can_advance_phase(),
"phases_completed": sorted(self.phases_completed),
"phase_history": dict(self.phase_history),
}
|