Spaces:
Sleeping
Sleeping
| """ | |
| Action utility helpers for safety and normalization. | |
| These helpers keep action generation robust across: | |
| - model output parsing | |
| - heuristic controllers | |
| - API payload conversion | |
| """ | |
| from __future__ import annotations | |
| from typing import Any, Optional, Tuple | |
| from models.schemas import GridAction | |
| def normalize_action_components( | |
| renewable_ratio: float, | |
| fossil_ratio: float, | |
| battery_action: float, | |
| ndigits: int = 3, | |
| ) -> tuple[float, float, float]: | |
| """Clamp + normalize action components and keep sum <= 1 after rounding.""" | |
| ren = max(0.0, min(1.0, float(renewable_ratio))) | |
| fos = max(0.0, min(1.0, float(fossil_ratio))) | |
| bat = max(-1.0, min(1.0, float(battery_action))) | |
| total = ren + fos | |
| if total > 1.0 and total > 0: | |
| ren /= total | |
| fos /= total | |
| # Round for consistent logging/UI while preserving constraints. | |
| ren = round(ren, ndigits) | |
| fos = round(fos, ndigits) | |
| bat = round(bat, ndigits) | |
| total_rounded = ren + fos | |
| if total_rounded > 1.0: | |
| overflow = round(total_rounded - 1.0, ndigits + 2) | |
| # Remove overflow from fossil first, then renewable. | |
| reduce_fos = min(fos, overflow) | |
| fos = round(fos - reduce_fos, ndigits) | |
| overflow = round(overflow - reduce_fos, ndigits + 2) | |
| if overflow > 0: | |
| ren = round(max(0.0, ren - overflow), ndigits) | |
| return ren, fos, bat | |
| def safe_grid_action( | |
| renewable_ratio: float, | |
| fossil_ratio: float, | |
| battery_action: float, | |
| ndigits: int = 3, | |
| ) -> GridAction: | |
| """Build a validated GridAction after normalization.""" | |
| ren, fos, bat = normalize_action_components( | |
| renewable_ratio=renewable_ratio, | |
| fossil_ratio=fossil_ratio, | |
| battery_action=battery_action, | |
| ndigits=ndigits, | |
| ) | |
| return GridAction( | |
| renewable_ratio=ren, | |
| fossil_ratio=fos, | |
| battery_action=bat, | |
| ) | |
| def coerce_grid_action( | |
| action_like: Any, | |
| default_action: Optional[GridAction] = None, | |
| ) -> Tuple[GridAction, Optional[str]]: | |
| """Convert action payloads to a valid GridAction with graceful fallback.""" | |
| try: | |
| if isinstance(action_like, GridAction): | |
| return action_like, None | |
| if isinstance(action_like, dict): | |
| payload = action_like.get("action", action_like) | |
| return safe_grid_action( | |
| renewable_ratio=payload.get("renewable_ratio", 0.5), | |
| fossil_ratio=payload.get("fossil_ratio", 0.5), | |
| battery_action=payload.get("battery_action", 0.0), | |
| ), None | |
| except Exception as exc: | |
| if default_action is not None: | |
| return default_action, f"invalid_action_payload: {type(exc).__name__}" | |
| raise | |
| if default_action is not None: | |
| return default_action, "invalid_action_type" | |
| raise ValueError("Unable to coerce action payload into GridAction.") | |