| 1 | """Stepwise director auto-generate flag, orthogonal to session ``source``.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from typing import Any |
| 6 | |
| 7 | SESSION_AUTO_GENERATE_KEY = "auto_generate" |
| 8 | DEFAULT_AUTO_GENERATE_DURATION_SEC = 30 |
| 9 | |
| 10 | |
| 11 | def coerce_bool(value: Any) -> bool | None: |
| 12 | """Parse wire/session booleans. Returns None when the value is absent/unknown.""" |
| 13 | if value is None: |
| 14 | return None |
| 15 | if isinstance(value, bool): |
| 16 | return value |
| 17 | if isinstance(value, (int, float)) and value in {0, 1}: |
| 18 | return bool(value) |
| 19 | if isinstance(value, str): |
| 20 | raw = value.strip().lower() |
| 21 | if raw in {"1", "true", "yes", "on"}: |
| 22 | return True |
| 23 | if raw in {"0", "false", "no", "off", ""}: |
| 24 | return False |
| 25 | return None |
| 26 | |
| 27 | |
| 28 | def resolve_auto_generate_from_wire(data: dict[str, Any] | None) -> bool | None: |
| 29 | """Read ``autoGenerate`` / ``auto_generate`` from a WS envelope or HTTP body.""" |
| 30 | if not isinstance(data, dict): |
| 31 | return None |
| 32 | if "autoGenerate" in data: |
| 33 | return coerce_bool(data.get("autoGenerate")) |
| 34 | if "auto_generate" in data: |
| 35 | return coerce_bool(data.get("auto_generate")) |
| 36 | return None |
| 37 | |
| 38 | |
| 39 | def get_auto_generate(metadata: dict[str, Any] | None) -> bool: |
| 40 | if not isinstance(metadata, dict): |
| 41 | return False |
| 42 | parsed = coerce_bool(metadata.get(SESSION_AUTO_GENERATE_KEY)) |
| 43 | return bool(parsed) |
| 44 | |
| 45 | |
| 46 | def apply_auto_generate(metadata: dict[str, Any], enabled: bool | None) -> bool: |
| 47 | """Set ``auto_generate`` on session metadata. Returns True when changed.""" |
| 48 | if not isinstance(metadata, dict) or enabled is None: |
| 49 | return False |
| 50 | current = get_auto_generate(metadata) |
| 51 | if current == enabled and SESSION_AUTO_GENERATE_KEY in metadata: |
| 52 | return False |
| 53 | metadata[SESSION_AUTO_GENERATE_KEY] = bool(enabled) |
| 54 | return True |
| 55 | |
| 56 | |
| 57 | def resolve_duration_sec_from_wire(data: dict[str, Any] | None) -> Any: |
| 58 | if not isinstance(data, dict): |
| 59 | return None |
| 60 | if "durationSec" in data: |
| 61 | return data.get("durationSec") |
| 62 | if "duration_sec" in data: |
| 63 | return data.get("duration_sec") |
| 64 | return None |
| 65 | |
| 66 | |
| 67 | def shot_count_for_auto_generate(duration_sec: Any) -> int: |
| 68 | """Map whole-video duration to shot count. Reject unknown tiers.""" |
| 69 | from nanobot.session.generation_settings import VALID_DURATIONS, duration_to_n_shots |
| 70 | |
| 71 | try: |
| 72 | parsed = int(duration_sec) |
| 73 | except (TypeError, ValueError) as exc: |
| 74 | raise ValueError(f"invalid auto_generate duration_sec={duration_sec!r}") from exc |
| 75 | mapped = duration_to_n_shots(parsed) |
| 76 | if mapped is None: |
| 77 | raise ValueError( |
| 78 | f"unsupported auto_generate duration_sec={parsed}; " |
| 79 | f"valid={sorted(VALID_DURATIONS)}" |
| 80 | ) |
| 81 | return mapped |
| 82 | |
| 83 | |
| 84 | def default_auto_generate_shot_count() -> int: |
| 85 | return shot_count_for_auto_generate(DEFAULT_AUTO_GENERATE_DURATION_SEC) |
| 86 | |
| 87 | |
| 88 | def locked_shot_count_from_goal(goal: Any) -> int | None: |
| 89 | """Return a positive locked ``goal.shot_count``, or None when unset/invalid.""" |
| 90 | if not isinstance(goal, dict): |
| 91 | return None |
| 92 | raw = goal.get("shot_count") |
| 93 | try: |
| 94 | parsed = int(raw) if raw not in (None, "") else 0 |
| 95 | except (TypeError, ValueError): |
| 96 | return None |
| 97 | return parsed if parsed > 0 else None |
| 98 | |
| 99 | |
| 100 | def locked_shot_count_from_state(state: Any) -> int | None: |
| 101 | if not isinstance(state, dict): |
| 102 | return None |
| 103 | return locked_shot_count_from_goal(state.get("goal")) |
| 104 | |
| 105 | |
| 106 | def session_n_shots(metadata: dict[str, Any] | None) -> int | None: |
| 107 | """Read the raw session ``n_shots`` value without applying defaults.""" |
| 108 | if not isinstance(metadata, dict): |
| 109 | return None |
| 110 | from nanobot.session.generation_settings import SESSION_NSHOT_KEY |
| 111 | |
| 112 | raw = metadata.get(SESSION_NSHOT_KEY) |
| 113 | try: |
| 114 | parsed = int(raw) if raw not in (None, "") else 0 |
| 115 | except (TypeError, ValueError): |
| 116 | return None |
| 117 | return parsed if parsed > 0 else None |
| 118 | |
| 119 | |
| 120 | def effective_auto_generate_shot_count( |
| 121 | *, |
| 122 | goal: Any = None, |
| 123 | state: Any = None, |
| 124 | metadata: dict[str, Any] | None = None, |
| 125 | ) -> int | None: |
| 126 | """Prefer the locked workplace shot count over session generation settings.""" |
| 127 | locked = locked_shot_count_from_goal(goal) |
| 128 | if locked is None: |
| 129 | locked = locked_shot_count_from_state(state) |
| 130 | if locked is not None: |
| 131 | return locked |
| 132 | return session_n_shots(metadata) |
| 133 |