Function bodies 99 total

class MadrixClient:
    config: MadrixConfig

    def _request(self, params: dict[str, Any]) -> httpx.Response:
        with httpx.Client(timeout=self.config.timeout_seconds) as client:
            response = client.get(self.config.base_url, params=params)
            response.raise_for_status()
            return response

    def health_check(self) -> dict[str, Any]:
        try:
            response = self._request({"GetProduct": ""})
            return {
                "reachable": True,
                "host": self.config.host,
                "port": self.config.http_port,
                "remote_path": self.config.remote_path,
                "status_code": response.status_code,
                "body_preview": response.text[:120],
            }
        except httpx.HTTPError as exc:
            return {
                "reachable": False,
                "host": self.config.host,
                "port": self.config.http_port,
                "remote_path": self.config.remote_pa

    def _request(self, params: dict[str, Any]) -> httpx.Response:
        with httpx.Client(timeout=self.config.timeout_seconds) as client:
            response = client.get(self.config.base_url, params=params)
            response.raise_for_status()
            return response

    def health_check(self) -> dict[str, Any]:
        try:
            response = self._request({"GetProduct": ""})
            return {
                "reachable": True,
                "host": self.config.host,
                "port": self.config.http_port,
                "remote_path": self.config.remote_path,
                "status_code": response.status_code,
                "body_preview": response.text[:120],
            }
        except httpx.HTTPError as exc:
            return {
                "reachable": False,
                "host": self.config.host,
                "port": self.config.http_port,
                "remote_path": self.config.remote_path,
                "error": str(exc),
            }

    def get_function(self, name: str, value: str | None = None) -> dict[str, Any]:
        params = {name: "" if value is None else str(value)}
        response = self._request(params)
        return {
            "function": name,
            "value": value,
            "status_code": response.status_code,
            "text": response.text,
            "lines": response.text.splitlines(),
            "url": str(response.request.url),
        }

    def set_function(self, name: str, value: str | int | float) -> dict[str, Any]:
        params = {name: str(value)}
        response = self._request(params)
        return {
            "function": name,
            "value": value,
            "status_code": response.status_code,
            "text": response.text,
            "url": str(response.request.url),
        }

    def get_functions(self, names: list[tuple[str, str | None]]) -> dict[str, Any]:
        params = [(name, "" if value is None else str(value)) for name, value in names]
        response = self._request(params)
        lines = response.text.splitlines()
        return {
            "functions": [name for name, _value in names],
            "status_code": response.status_code,
            "lines": lines,
            "url": str(response.request.url),
        }

def _parse_port(env_name: str, default: str) -> int:
    raw = os.getenv(env_name, default)
    try:
        port = int(raw)
    except ValueError:
        raise ValueError(f"{env_name}={raw!r} is not a valid integer") from None
    if not (1 <= port <= 65535):
        raise ValueError(f"{env_name}={port} is outside valid port range 1-65535")
    return port

def _parse_allowed_hosts(raw: str) -> frozenset[str]:
    hosts = frozenset(h.strip() for h in raw.split(",") if h.strip())
    if not hosts:
        raise ValueError("MADRIX_ALLOWED_HOSTS must contain at least one host")
    return hosts

def _normalize_remote_path(raw: str) -> str:
    path = (raw or "/RemoteCommands/").strip()
    if not path.startswith("/"):
        path = "/" + path
    if not path.endswith("/"):
        path += "/"
    return path

class MadrixConfig:
    host: str = "127.0.0.1"
    http_port: int = 80
    remote_path: str = "/RemoteCommands/"
    allowed_hosts: frozenset[str] = field(default_factory=lambda: frozenset({"127.0.0.1", "localhost", "::1"}))
    safety_profile: str = "lab"
    read_only: bool = False
    confirm_destructive: bool = False
    timeout_seconds: float = 5.0

    @property
    def base_url(self) -> str:
        return f"http://{self.host}:{self.http_port}{self.remote_path}"

    def check_host_allowed(self) -> None:
        if "*" in self.allowed_hosts:
            return
        if self.host not in self.allowed_hosts:
            raise ValueError(
                f"Host {self.host!r} is not in MADRIX_ALLOWED_HOSTS. "
                f"Allowed: {', '.join(sorted(self.allowed_hosts))}. "
                f"Set MADRIX_ALLOWED_HOSTS=* to allow any host."
            )

    def check_host_allowed(self) -> None:
        if "*" in self.allowed_hosts:
            return
        if self.host not in self.allowed_hosts:
            raise ValueError(
                f"Host {self.host!r} is not in MADRIX_ALLOWED_HOSTS. "
                f"Allowed: {', '.join(sorted(self.allowed_hosts))}. "
                f"Set MADRIX_ALLOWED_HOSTS=* to allow any host."
            )

def _load_safety_profile() -> tuple[str, bool, bool]:
    profile = os.getenv("MADRIX_SAFETY_PROFILE", "lab").strip().lower() or "lab"
    if profile not in _SAFETY_PROFILES:
        raise ValueError(
            f"MADRIX_SAFETY_PROFILE={profile!r} is invalid. "
            f"Choose one of: {', '.join(sorted(_SAFETY_PROFILES))}."
        )
    defaults = {
        "lab": {"read_only": False, "confirm_destructive": False},
        "show-safe": {"read_only": False, "confirm_destructive": True},
        "read-only": {"read_only": True, "confirm_destructive": True},
    }[profile]
    read_only = _parse_bool(os.getenv("MADRIX_READ_ONLY", "1" if defaults["read_only"] else "0"))
    confirm_destructive = _parse_bool(
        os.getenv("MADRIX_CONFIRM_DESTRUCTIVE", "1" if defaults["confirm_destructive"] else "0")
    )
    return profile, read_only, confirm_destructive

def load_config() -> MadrixConfig:
    profile, read_only, confirm_destructive = _load_safety_profile()
    config = MadrixConfig(
        host=os.getenv("MADRIX_HOST", "127.0.0.1"),
        http_port=_parse_port("MADRIX_HTTP_PORT", "80"),
        remote_path=_normalize_remote_path(os.getenv("MADRIX_REMOTE_PATH", "/RemoteCommands/")),
        allowed_hosts=_parse_allowed_hosts(os.getenv("MADRIX_ALLOWED_HOSTS", _DEFAULT_ALLOWED_HOSTS)),
        safety_profile=profile,
        read_only=read_only,
        confirm_destructive=confirm_destructive,
        timeout_seconds=float(os.getenv("MADRIX_TIMEOUT_SECONDS", "5.0")),
    )
    config.check_host_allowed()
    return config

def _handle_errors(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except json.JSONDecodeError as exc:
            return _error(f"Invalid JSON input: {exc.msg}", blocked=True)
        except ValueError as exc:
            return _error(str(exc), blocked=True)
        except Exception as exc:
            return _error(f"Unexpected error: {type(exc).__name__}: {exc}", blocked=False)
    return wrapper

def _check_write() -> None:
    if load_config().read_only:
        raise ValueError("Server is in read-only mode (MADRIX_READ_ONLY=1). Write operations are disabled.")

def _check_destructive(function_name: str, confirm: bool) -> None:
    _check_write()
    config = load_config()
    if config.confirm_destructive and function_name in _DESTRUCTIVE_SET_FUNCTIONS and not confirm:
        raise ValueError(
            f"Destructive operation {function_name!r} requires confirm=true "
            f"(MADRIX_CONFIRM_DESTRUCTIVE is enabled)."
        )

def _validate_u8(value: int, name: str) -> None:
    if not (0 <= value <= 255):
        raise ValueError(f"{name} must be between 0 and 255, got {value}")

def _validate_non_negative_int(value: int, name: str) -> None:
    if value < 0:
        raise ValueError(f"{name} must be >= 0, got {value}")

def _validate_int_range(value: int, name: str, *, minimum: int, maximum: int) -> None:
    if not (minimum <= value <= maximum):
        raise ValueError(f"{name} must be between {minimum} and {maximum}, got {value}")

def _validate_string_non_empty(value: str, name: str) -> None:
    if not value.strip():
        raise ValueError(f"{name} must be a non-empty string.")

def _normalize_deck(deck: str) -> str:
    normalized = deck.strip().upper()
    if normalized not in _DECK_SUFFIXES:
        raise ValueError(f"deck must be one of {sorted(_DECK_SUFFIXES)}, got {deck!r}")
    return normalized

def _set_no_value(function_name: str, *, confirm: bool = False) -> dict[str, Any]:
    _check_destructive(function_name, confirm)
    return _client().set_function(function_name, "")

def _set_bool(function_name: str, enabled: bool, *, confirm: bool = False) -> dict[str, Any]:
    _check_destructive(function_name, confirm)
    return _client().set_function(function_name, 1 if enabled else 0)

def _set_u8(function_name: str, value: int, field_name: str = "value", *, confirm: bool = False) -> dict[str, Any]:
    _validate_u8(value, field_name)
    _check_destructive(function_name, confirm)
    return _client().set_function(function_name, value)

def _set_non_negative(function_name: str, value: int, field_name: str = "value", *, confirm: bool = False) -> dict[str, Any]:
    _validate_non_negative_int(value, field_name)
    _check_destructive(function_name, confirm)
    return _client().set_function(function_name, value)

def _set_float_min(function_name: str, value: float, field_name: str, minimum: float, *, confirm: bool = False) -> dict[str, Any]:
    if value < minimum:
        raise ValueError(f"{field_name} must be >= {minimum}, got {value}")
    _check_destructive(function_name, confirm)
    return _client().set_function(function_name, value)

def _set_pair_int(function_name: str, first_name: str, first_value: int, second_name: str, second_value: int, *,
                  second_validator=None, first_validator=None, confirm: bool = False) -> dict[str, Any]:
    if first_validator:
        first_validator(first_value)
    else:
        _validate_non_negative_int(first_value, first_name)
    if second_validator:
        second_validator(second_value)
    else:
        _validate_non_negative_int(second_value, second_name)
    _check_destructive(function_name, confirm)
    return _client().set_function(function_name, _underscore_pair(first_value, second_value))

def _set_pair_string(function_name: str, first_name: str, first_value: int, second_name: str, second_value: str, *,
                     first_validator=None, confirm: bool = False) -> dict[str, Any]:
    if first_validator:
        first_validator(first_value)
    else:
        _validate_non_negative_int(first_value, first_name)
    _validate_string_non_empty(second_value, second_name)
    _check_destructive(function_name, confirm)
    return _client().set_function(function_name, _underscore_pair(first_value, second_value))

def get_server_config() -> str:
    config = load_config()
    return _json({
        "host": config.host,
        "http_port": config.http_port,
        "remote_path": config.remote_path,
        "base_url": config.base_url,
        "allowed_hosts": sorted(config.allowed_hosts),
        "safety_profile": config.safety_profile,
        "read_only": config.read_only,
        "confirm_destructive": config.confirm_destructive,
    })

def send_http_get_function(function_name: str, value: str = "") -> str:
    if not function_name.startswith("Get"):
        raise ValueError("function_name must start with 'Get'.")
    return _json(_client().get_function(function_name, value if value else None))

def send_http_set_function(function_name: str, value: str, confirm: bool = False) -> str:
    if not function_name.startswith("Set"):
        raise ValueError("function_name must start with 'Set'.")
    _check_destructive(function_name, confirm)
    return _json(_client().set_function(function_name, value))

def get_storage_state() -> str:
    client = _client()
    result = client.get_functions(
        [
            ("GetStorage", None),
            ("GetStorageDeckA", None),
            ("GetStorageDeckB", None),
            ("GetStoragePlace", None),
            ("GetStoragePlaceDeckA", None),
            ("GetStoragePlaceDeckB", None),
        ]
    )
    lines = result["lines"]
    return _json({
        "storage": lines[0] if len(lines) > 0 else None,
        "storage_deck_a": lines[1] if len(lines) > 1 else None,
        "storage_deck_b": lines[2] if len(lines) > 2 else None,
        "place": lines[3] if len(lines) > 3 else None,
        "place_deck_a": lines[4] if len(lines) > 4 else None,
        "place_deck_b": lines[5] if len(lines) > 5 else None,
        "raw": result,
    })

def get_audio_state() -> str:
    client = _client()
    result = client.get_functions(
        [
            ("GetAudioInputMute", None),
            ("GetAudioOutputMute", None),
            ("GetAudioInputFader", None),
            ("GetAudioOutputFader", None),
            ("GetAutoGainControl", None),
            ("GetSoundLevel", "0"),
            ("GetSoundLevel", "1"),
        ]
    )
    lines = result["lines"]
    return _json({
        "input_mute": lines[0] if len(lines) > 0 else None,
        "output_mute": lines[1] if len(lines) > 1 else None,
        "input_fader": lines[2] if len(lines) > 2 else None,
        "output_fader": lines[3] if len(lines) > 3 else None,
        "auto_gain_input": lines[4] if len(lines) > 4 else None,
        "sound_level_left": lines[5] if len(lines) > 5 else None,
        "sound_level_right": lines[6] if len(lines) > 6 else None,
        "raw": result,
    })

def get_output_state() -> str:
    client = _client()
    result = client.get_functions(
        [
            ("GetOutputFreeze", None),
            ("GetOutputStrobe", None),
            ("GetOutputStrobeValue", None),
            ("GetOutputStrobeColorRed", None),
            ("GetOutputStrobeColorGreen", None),
            ("GetOutputStrobeColorBlue", None),
            ("GetOutputStrobeColorWhite", None),
            ("GetOutputFilter", None),
            ("GetOutputFilterRed", None),
            ("GetOutputFilterGreen", None),
            ("GetOutputFilterBlue", None),
            ("GetOutputFilterWhite", None),
        ]
    )
    lines = result["lines"]
    return _json({
        "freeze": lines[0] if len(lines) > 0 else None,
        "strobe": lines[1] if len(lines) > 1 else None,
        "strobe_value": lines[2] if len(lines) > 2 else None,
        "strobe_red": lines[3] if len(lines) > 3 else None,
        "strobe_green": lines[4] if len(lines) > 4 else None,
        "strobe

def get_cuelist_summary() -> str:
    client = _client()
    result = client.get_functions(
        [
            ("GetCuelistSelected", None),
            ("GetCuelistPlay", None),
            ("GetCuelistCount", None),
            ("GetCuelistCueCurrent", None),
            ("GetCuelistTimeCodeSource", None),
        ]
    )
    lines = result["lines"]
    return _json({
        "selected": lines[0] if len(lines) > 0 else None,
        "playing": lines[1] if len(lines) > 1 else None,
        "count": lines[2] if len(lines) > 2 else None,
        "current_cue": lines[3] if len(lines) > 3 else None,
        "timecode_source": lines[4] if len(lines) > 4 else None,
        "raw": result,
    })

def set_fade_time(seconds: float) -> str:
    _check_write()
    if seconds < 0:
        raise ValueError(f"seconds must be >= 0, got {seconds}")
    return _json(_client().set_function("SetFadeTime", seconds))

def set_fade_type(name: str) -> str:
    _check_write()
    if name not in _FADE_TYPES:
        raise ValueError(f"name must be one of {sorted(_FADE_TYPES)}, got {name!r}")
    return _json(_client().set_function("SetFadeType", name))

def set_fade_value(value: int) -> str:
    _check_write()
    _validate_u8(value, "value")
    return _json(_client().set_function("SetFadeValue", value))

def trigger_fade_auto() -> str:
    _check_write()
    return _json(_set_no_value("SetFadeAuto"))

def trigger_fade_to_deck_a() -> str:
    _check_write()
    return _json(_set_no_value("SetFadeToDeckA"))

def trigger_fade_to_deck_b() -> str:
    _check_write()
    return _json(_set_no_value("SetFadeToDeckB"))

def trigger_fade_to_center() -> str:
    _check_write()
    return _json(_set_no_value("SetFadeToCenter"))

def set_master(level: int, confirm: bool = False) -> str:
    _validate_u8(level, "level")
    _check_destructive("SetMaster", confirm)
    return _json(_client().set_function("SetMaster", level))

def set_blackout(enabled: bool, confirm: bool = False) -> str:
    _check_destructive("SetBlackout", confirm)
    return _json(_client().set_function("SetBlackout", 1 if enabled else 0))

def set_storage(index: int, confirm: bool = False) -> str:
    _validate_non_negative_int(index, "index")
    _check_destructive("SetStorage", confirm)
    return _json(_client().set_function("SetStorage", index))

def set_storage_deck_a(index: int, confirm: bool = False) -> str:
    _validate_non_negative_int(index, "index")
    _check_destructive("SetStorageDeckA", confirm)
    return _json(_client().set_function("SetStorageDeckA", index))

def set_storage_deck_b(index: int, confirm: bool = False) -> str:
    _validate_non_negative_int(index, "index")
    _check_destructive("SetStorageDeckB", confirm)
    return _json(_client().set_function("SetStorageDeckB", index))

def set_storage_without_fade(index: int, confirm: bool = False) -> str:
    _validate_non_negative_int(index, "index")
    _check_destructive("SetStorageWithoutFade", confirm)
    return _json(_client().set_function("SetStorageWithoutFade", index))

def set_storage_without_fade_deck_a(index: int, confirm: bool = False) -> str:
    _validate_non_negative_int(index, "index")
    _check_destructive("SetStorageWithoutFadeDeckA", confirm)
    return _json(_client().set_function("SetStorageWithoutFadeDeckA", index))

def set_storage_without_fade_deck_b(index: int, confirm: bool = False) -> str:
    _validate_non_negative_int(index, "index")
    _check_destructive("SetStorageWithoutFadeDeckB", confirm)
    return _json(_client().set_function("SetStorageWithoutFadeDeckB", index))