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 []