Extract index switch policy

index_switch_policy.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+MAX_SWITCH_INPUTS = 64
+INDEX_SWITCH_MODES = ["pick_input", "route_output"]
+INDEX_SWITCH_BASES = ["one_based", "zero_based"]
+INDEX_SWITCH_MISSING_BEHAVIORS = ["fallback", "none", "clamp", "wrap"]
+
+
+def normalize_index_base(value: Any) -> str:
+    return value if value in INDEX_SWITCH_BASES else "one_based"
+
+
+def normalize_missing_behavior(value: Any) -> str:
+    return value if value in INDEX_SWITCH_MISSING_BEHAVIORS else "fallback"
+
+
+def normalize_mode(value: Any) -> str:
+    return value if value in INDEX_SWITCH_MODES else "pick_input"
+
+
+def available_input_indices(kwargs: dict[str, Any]) -> list[int]:
+    indices = []
+    for key in kwargs:
+        match = re.match(r"^input_(\d+)$", str(key))
+        if match:
+            indices.append(int(match.group(1)))
+    return sorted(set(indices))
+
+
+def requested_index(index: Any, index_base: str) -> int:
+    requested = int(index)
+    return requested + 1 if normalize_index_base(index_base) == "zero_based" else requested
+
+
+def resolved_input_index(requested: int, available: list[int], missing_behavior: str) -> int | None:
+    missing_behavior = normalize_missing_behavior(missing_behavior)
+    if requested in available:
+        return requested
+    if missing_behavior in ("fallback", "none") or not available:
+        return None
+    if missing_behavior == "wrap":
+        return available[(requested - 1) % len(available)]
+    if requested <= available[0]:
+        return available[0]
+    if requested >= available[-1]:
+        return available[-1]
+    lower = [value for value in available if value <= requested]
+    return lower[-1] if lower else available[0]
+
+
+def input_selection(index: Any, index_base: str, missing_behavior: str, kwargs: dict[str, Any]) -> tuple[int, int | None, list[int]]:
+    requested = requested_index(index, index_base)
+    available = available_input_indices(kwargs)
+    selected = resolved_input_index(requested, available, missing_behavior)
+    return requested, selected, available
+
+
+def route_selection(index: Any, index_base: str, missing_behavior: str, max_outputs: int = MAX_SWITCH_INPUTS) -> tuple[int, int | None]:
+    requested = requested_index(index, index_base)
+    max_outputs = max(1, int(max_outputs))
+    missing_behavior = normalize_missing_behavior(missing_behavior)
+    if 1 <= requested <= max_outputs:
+        return requested, requested
+    if missing_behavior == "wrap":
+        return requested, ((requested - 1) % max_outputs) + 1
+    if missing_behavior == "clamp":
+        return requested, min(max(requested, 1), max_outputs)
+    return requested, None
+
+
+def input_status(requested: int, selected: int | None, used_fallback: bool, available: list[int]) -> str:
+    available_text = ",".join(str(index) for index in available) or "none"
+    if used_fallback:
+        return f"requested=input_{requested}; selected=fallback; available={available_text}"
+    if selected is None:
+        return f"requested=input_{requested}; selected=none; available={available_text}"
+    return f"requested=input_{requested}; selected=input_{selected}; available={available_text}"
+
+
+def route_status(requested: int, selected: int | None, max_outputs: int = MAX_SWITCH_INPUTS) -> str:
+    selected_text = "none" if selected is None else f"output_{selected}"
+    return f"requested=output_{requested}; selected={selected_text}; range=1-{max_outputs}"
+
+
+def lazy_inputs(index: Any, mode: str, index_base: str, missing_behavior: str, kwargs: dict[str, Any]) -> list[str]:
+    mode = normalize_mode(mode)
+    missing_behavior = normalize_missing_behavior(missing_behavior)
+    if mode == "route_output":
+        return ["route_value"] if "route_value" in kwargs else []
+    requested, selected, _available = input_selection(index, index_base, missing_behavior, kwargs)
+    selected_name = f"input_{selected}" if selected is not None else f"input_{requested}"
+    if selected_name in kwargs:
+        return [selected_name]
+    if missing_behavior == "fallback" and "fallback" in kwargs:
+        return ["fallback"]
+    return []