Build a fresh orchestration state from an optimization plan.
(plan: dict)
| 134 | |
| 135 | |
| 136 | def _initialize_state_from_plan(plan: dict) -> dict: |
| 137 | """Build a fresh orchestration state from an optimization plan.""" |
| 138 | kernels_raw = plan.get("kernels_to_optimize", plan.get("kernels", [])) |
| 139 | kernels = [] |
| 140 | for i, kp in enumerate(kernels_raw): |
| 141 | kernels.append( |
| 142 | _default_kernel_entry( |
| 143 | rank=kp.get("rank", i + 1), |
| 144 | file=kp.get("file", f"workspace/kernel_{kp.get('op_type', 'unknown')}_{i+1}.py"), |
| 145 | op_type=kp.get("op_type", "unknown"), |
| 146 | pct_total=kp.get("pct_total", 0.0), |
| 147 | ) |
| 148 | ) |
| 149 | |
| 150 | first_file = kernels[0]["file"] if kernels else None |
| 151 | if kernels: |
| 152 | kernels[0]["status"] = STATUS_OPTIMIZING |
| 153 | |
| 154 | return { |
| 155 | "current_kernel_idx": 0, |
| 156 | "current_kernel_file": first_file, |
| 157 | "started_at": _now_iso(), |
| 158 | "kernels": kernels, |
| 159 | } |
| 160 | |
| 161 | |
| 162 | def load_state() -> dict | None: |
no test coverage detected