AppSettings

def __init__(self, setting: List[Dict[str, Any]]) -> None:
    """Parse the command line and environment against `setting`.

    Unrecognized arguments are tolerated so a wrapper can pass its own
    through, which also means a misspelled flag is accepted and ignored
    rather than rejected; warning about each by name is what makes that
    visible, since a setting that never took its value looks the same as
    one left at its default.
    """
    parser = argparse.ArgumentParser(
        description="Selkies WebSocket Streaming Server"
    )
    self._setting_definitions = setting
    self._add_arguments(parser)
    args, unknown = parser.parse_known_args()
    for token in unknown:
        if token.startswith("-"):
            logging.warning(
                "Ignoring unrecognized argument %s", token.split("=", 1)[0]
            )
    self._process_and_set_attributes(args)
    self._post_process_settings()

@staticmethod
def _fallback_env_vars(setting: Dict[str, Any]) -> Tuple[str, ...]:
    """Return a setting's fallback env aliases as a tuple: `env_var` may be
    one name or an ordered list of names (earlier entries win)."""
    fallback = setting.get("env_var")
    if not fallback:
        return ()
    if isinstance(fallback, str):
        return (fallback,)
    return tuple(fallback)

def _add_arguments(self, parser: argparse.ArgumentParser) -> None:
    """Add one string-typed CLI argument per setting definition.

    Every flag parses as a raw string (type conversion happens later in
    `_process_and_set_attributes`) so CLI and environment values flow
    through the identical parsing path. Both `--my-setting` and
    `--my_setting` are registered: dashes are the documented spelling, but
    the setting's own name is what every environment variable uses, so
    the underscore form is accepted rather than dropped as unknown.
    """
    for setting in self._setting_definitions:
        name = setting["name"]
        cli_flags = [f"--{name.replace('_', '-')}"]
        if "_" in name:
            cli_flags.append(f"--{name}")
        standard_env_var = f"SELKIES_{name.upper()}"
        fallback_env_vars = self._fallback_env_vars(setting)
        env_help_text = f"Env: {standard_env_var}"
        if fallback_env_vars:
            env_help_text = f"Env: {standard_env_var} (or {', '.join(fallback_env_vars)})"
        parser.add_argument(
            *cli_flags,
            dest=name,
            type=str,
            default=None,
            help=f"{setting['help']} ({env_help_text})",
        )

def _process_and_set_attributes(self, args: argparse.Namespace) -> None:
    """Resolve every setting's raw value, parse it by type, and set it as
    an instance attribute.

    Also records which settings were explicitly overridden (CLI or env) in
    `self._overridden`. An override that resolves to the built-in value
    (the empty string, or an entirely-invalid enum/list) reads as
    not-provided too: an operator intent the server discarded must not go
    on suppressing client changes or the client's own conditional default.

    Enum and list items match `allowed` case-insensitively and are carried
    forward in its spelling, so `SELKIES_ENCODER=JPEG` selects the encoder
    every consumer compares against. A numeric enum declaring
    `meta.value_range` takes a single in-range value verbatim as the server
    value while `allowed` keeps the curated stops for the web UI: the
    server accepts more than the UI offers.

    Manual-resolution mode follows a positive manual width or height
    override, or an explicit `manual_resolution`: 0 is both the
    built-in default and the no-manual-width sentinel, so a templated
    launcher emitting `--manual-width 0` for an unset field must not lock
    the display to 1024x768.
    """
    processed = {}
    overrides = {}
    for setting in self._setting_definitions:
        name = setting["name"]
        stype = setting["type"]
        cli_val = getattr(args, name, None)
        std_env_val = os.environ.get(f"SELKIES_{name.upper()}")
        fallback_env_val = None
        for fallback_var in self._fallback_env_vars(setting):
            fallback_env_val = os.environ.get(fallback_var)
            if fallback_env_val is not None:
                break
        is_override = (
            cli_val is not None
            or std_env_val is not None
            or fallback_env_val is not None
        )
        overrides[name] = is_override

        raw_value = (
            cli_val
            if cli_val is not None
            else (
                std_env_val
                if std_env_val is not None
                else (
                    fallback_env_val
                    if fallback_env_val is not None
                    else setting["default"]
                )
            )
        )
        if (
            is_override
            and stype != "list"
            and str(raw_value).strip() == ""
        ):
            is_override = False
            overrides[name] = False
            raw_value = setting["default"]
        processed_value = None
        try:
            if stype == "bool":
                suffixes = [
                    part.strip()
                    for part in str(raw_value).strip().lower().split("|")[1:]
                ]
                processed_value = (parse_bool(raw_value), "locked" in suffixes)
            elif stype in ["enum", "list"]:
                if is_override:
                    master_list = setting.get("meta", {}).get("allowed", [])
                    raw_value_str = str(raw_value)
                    if stype == "list" and raw_value_str.strip().lower() in ("", "none"):
                        setting["meta"]["allowed"] = []
                        processed_value = []
                    else:
                        user_items = [item.strip() for item in raw_value_str.split(',') if item.strip()]
                        if name == "encoder":
                            user_items = [canonical_encoder(item) for item in user_items]
                        canonical = {item.lower(): item for item in master_list}
                        user_items = [canonical.get(item.lower(), item) for item in user_items]
                        valid_items = [item for item in user_items if item in master_list]
                        vr = setting.get("meta", {}).get("value_range")
                        in_range_value = None
                        # Gated on value_range, not on `not valid_items`: an in-range
                        # value equal to a curated stop must still keep the full menu.
                        if stype == "enum" and vr and len(user_items) == 1:
                            try:
                                n = float(user_items[0])
                                if vr[0] <= n <= vr[1]:
                                    # '128000.0' must not reach int() consumers.
                                    in_range_value = (
                                        str(int(n)) if n == int(n) else user_items[0]
                                    )
                            except ValueError:
                                pass
                        if in_range_value is not None:
                            processed_value = in_range_value
                        elif valid_items:
                            setting["meta"]["allowed"] = valid_items
                            processed_value = valid_items[0] if stype == "enum" else valid_items
                        else:
                            is_override = False
                            overrides[name] = False
                            if user_items:
                                logging.warning(
                                    f"Invalid value(s) '{raw_value_str}' for {name}; "
                                    f"keeping the full allowed set with the system default."
                                )
                            default_str = str(setting["default"])
                            default_items = [item.strip() for item in default_str.split(',') if item.strip() and item.strip() in master_list]
                            if stype == "enum":
                                processed_value = default_items[0] if default_items else setting["default"]
                            else:
                                processed_value = default_items
                else:
                    if stype == "enum":
                        processed_value = setting["default"]
                    else:
                        processed_value = [
                            item.strip()
                            for item in str(setting["default"]).split(",")
                            if item.strip()
                        ]
            elif stype in ("int", "float"):
                processed_value = int(raw_value) if stype == "int" else float(raw_value)
                # Only declared bounds clamp, so -1/negative sentinels survive.
                meta = setting.get("meta") or {}
                lo = setting.get("min", meta.get("min"))
                hi = setting.get("max", meta.get("max"))
                orig = processed_value
                if lo is not None:
                    processed_value = max(lo, processed_value)
                if hi is not None:
                    processed_value = min(hi, processed_value)
                if processed_value != orig:
                    logging.warning(
                        f"Setting '{name}' value {orig} out of range [{lo},{hi}], clamped to {processed_value}"
                    )
            elif stype == "str":
                processed_value = str(raw_value)
            elif stype == "range":
                tokens = [
                    token.strip()
                    for token in str(raw_value).split(",")
                    if token.strip()
                ]
                span = None
                initial = None
                for token in tokens:
                    span_match = re.fullmatch(
                        r"(-?\d+(?:\.\d+)?)\s*-\s*(-?\d+(?:\.\d+)?)", token
                    )
                    if span_match:
                        span = (
                            _range_number(span_match.group(1)),
                            _range_number(span_match.group(2)),
                        )
                    else:
                        initial = _range_number(token)
                if span is None and initial is None:
                    raise ValueError("no span or value given")
                meta = setting.get("meta")
                if span is None:
                    def_lo, def_hi = sorted(
                        _range_number(part)
                        for part in str(setting["default"]).split("-", 1)
                    )
                    processed_value = (
                        min(def_lo, initial),
                        max(def_hi, initial),
                    )
                    if meta is not None:
                        meta["default_value"] = initial
                else:
                    lo, hi = sorted(span)
                    processed_value = (lo, hi)
                    if meta is not None and initial is not None:
                        clamped = max(lo, min(initial, hi))
                        if clamped != initial:
                            logging.warning(
                                f"Setting '{name}' initial value {initial} "
                                f"outside span {lo}-{hi}, clamped to {clamped}"
                            )
                        meta["default_value"] = clamped
                    elif meta is not None and "default_value" in meta:
                        meta["default_value"] = max(
                            lo, min(meta["default_value"], hi)
                        )
        except (ValueError, TypeError, IndexError) as e:
            logging.error(
                f"Could not parse setting '{name}' with value '{raw_value}'. Using default. Error: {e}"
            )
            processed_value = setting["default"]
            if stype == "range":
                min_val, max_val = (
                    _range_number(part)
                    for part in str(processed_value).split("-", 1)
                )
                processed_value = (min_val, max_val)
        processed[name] = processed_value
    width_overridden = overrides.get("manual_width", False) and processed.get("manual_width", 0) > 0
    height_overridden = overrides.get("manual_height", False) and processed.get("manual_height", 0) > 0
    manual_mode_bool_is_set = processed.get(
        "manual_resolution", (False, False)
    )[0]
    should_be_in_manual_mode = (
        width_overridden or height_overridden or manual_mode_bool_is_set
    )
    if should_be_in_manual_mode:
        logging.info(
            "A manual resolution setting was activated; locking to manual mode."
        )
        processed["manual_resolution"] = (True, True)
        if processed.get("manual_width", 0) <= 0:
            processed["manual_width"] = 1024
            logging.info("Manual width not set or invalid, defaulting to 1024.")
        if processed.get("manual_height", 0) <= 0:
            processed["manual_height"] = 768
            logging.info("Manual height not set or invalid, defaulting to 768.")
    for name, value in processed.items():
        setattr(self, name, value)
    self._overridden = overrides

def was_provided(self, name: str) -> bool:
    """Whether a setting was given on the command line or in the environment.

    A setting left to its built-in default reads as not provided, and so does one
    overridden to the empty string, which carries the "use the default" meaning.
    """
    return bool(getattr(self, "_overridden", {}).get(name, False))

def on_software_h264_path(self) -> bool:
    """Whether the server's own defaults put a session on the software
    H.264 path: the striped encoder, or h264enc with software encoding
    forced by use_cpu or gpu_id=-1."""
    forced = bool(self.use_cpu[0]) or str(self.gpu_id).strip() == "-1"
    return software_h264_path(self.encoder, forced)

def resolve_rate_control_default(self) -> None:
    """Apply the transport's rate-control default for the current mode.

    WebRTC streams default to CBR whatever the encoder: a
    congestion-controlled transport needs the encoder holding a bandwidth
    target. Websockets streams are quality-driven (`ENCODER_RC_DEFAULTS`),
    except that OpenH264 — the software H.264 encoder of a GPL-free
    pixelflux build — targets a bandwidth, so a session known to be on the
    software path defaults to CBR; encoders not listed keep their value.
    The dashboards derive the same default client-side
    (conditional-settings.js) from the published `software_h264_encoder`.

    A no-op when the operator pinned rate_control_mode or disabled rate
    control. Called again on a live transport switch so an unpinned mode
    tracks the transport actually streaming.
    """
    if not self.enable_rate_control[0] or self.was_provided("rate_control_mode"):
        return
    if self.mode == "webrtc":
        self.rate_control_mode = "cbr"
    elif self.on_software_h264_path() and software_h264_encoder() == "openh264":
        self.rate_control_mode = "cbr"
    else:
        self.rate_control_mode = self.ENCODER_RC_DEFAULTS.get(
            self.encoder, self.rate_control_mode
        )
    self.resolve_paint_over_default()

def apply_webrtc_encoder_filter(self) -> None:
    """Bring the `encoder` knob — the published menu and the value — in
    line with the transport.

    One encoder knob drives both transports: websockets framing is what
    only websockets can carry, so in webrtc mode the menu and value are
    brought into the WebRTC-producible subset of the operator's menu and
    a choice that cannot stream falls back with a warning. The first call
    (from _post_process_settings, after any operator narrowing of
    `allowed`) snapshots that menu and value, so a live transport switch
    filters against the operator's menu rather than the shipped one, and
    a switch back to websockets restores both: neither a websockets-only
    capability (jpeg, h264enc-striped) nor an operator narrowing or pin
    is lost across a round trip.

    A clamped value is stashed so the switch back can also restore a
    client's websockets-only pick (clients write the session encoder
    through to this knob): the stash applies only while the clamp
    fallback is still in force and the client asserted nothing newer
    during the webrtc leg — a fresh pick, flagged through
    _encoder_client_set, always wins over the stash.
    """
    enc_definition = next(
        (s for s in self._setting_definitions if s["name"] == "encoder"),
        None,
    )
    if enc_definition is None:
        return
    if not hasattr(self, "_operator_encoder_allowed"):
        self._operator_encoder_allowed = list(enc_definition["meta"]["allowed"])
        self._operator_encoder_value = self.encoder
    if self.mode != "webrtc":
        enc_definition["meta"]["allowed"] = list(self._operator_encoder_allowed)
        if self.encoder not in self._operator_encoder_allowed:
            self.encoder = self._operator_encoder_value
        stash = getattr(self, "_pre_webrtc_encoder", None)
        if stash is not None:
            if (
                not getattr(self, "_encoder_client_set", False)
                and self.encoder == getattr(self, "_webrtc_encoder_fallback", None)
                and stash in self._operator_encoder_allowed
            ):
                self.encoder = stash
            self._pre_webrtc_encoder = None
            self._webrtc_encoder_fallback = None
        return
    allowed = [
        item for item in self._operator_encoder_allowed
        if item in WEBRTC_ENCODER_CHOICES
    ]
    if not allowed:
        allowed = list(WEBRTC_ENCODER_CHOICES)
    enc_definition["meta"]["allowed"] = allowed
    if self.encoder not in allowed:
        fallback = (
            enc_definition["default"]
            if enc_definition["default"] in allowed
            else allowed[0]
        )
        if self.was_provided("encoder"):
            logging.warning(
                "Encoder %r is not available for WebRTC (%s); using %r.",
                self.encoder,
                ", ".join(allowed),
                fallback,
            )
        self._pre_webrtc_encoder = self.encoder
        self._webrtc_encoder_fallback = fallback
        self._encoder_client_set = False
        self.encoder = fallback
    else:
        self._pre_webrtc_encoder = None
        self._webrtc_encoder_fallback = None

def resolve_paint_over_default(self) -> None:
    """Default paint-over off on a bandwidth-targeted stream (CBR): the
    static-scene repaint forces periodic bursts that a bitrate cap pays
    for in motion quality, and no client has asked for the trade yet.

    An explicit operator use_paint_over_quality choice wins via the
    override check inside; client choices live in per-display state and
    the dashboards' own precedence ladder, which this default never
    outranks. Called from anywhere rate control resolves.
    """
    if self.was_provided("use_paint_over_quality"):
        return
    self.use_paint_over_quality = (
        self.rate_control_mode != "cbr",
        self.use_paint_over_quality[1],
    )

def _post_process_settings(self) -> None:
    """Normalize and cross-check settings whose meaning spans several
    entries.

    The subfolder is stored the one way that composes — empty, or a
    leading slash and no trailing one — because every route concatenates
    it in front of a slash-led path; "/" is the root, which is the empty
    prefix. The transport mode is lowercased before anything branches on
    it: the service registry is keyed on this value, so a differently-cased
    one would abort the server at startup rather than select the transport
    it names. With rate control locked off the engine runs constant
    quality on both transports, so the resolved mode and the menu
    published to clients are CRF alone; an encoder-derived "cbr" would
    leave the dashboards showing a bitrate slider the encoder ignores and
    hiding the CRF slider in force. Microphone forwarding requires audio.
    The clipboard policy is normalized to exactly one of its four values.
    The TURN username (the REST service's x-auth-user and the HMAC
    credential alike) defaults to a generic name so the credential stays
    stable and non-empty (`<expiry>:selkies`) instead of a bare
    `<expiry>:` or a volatile pod hostname.
    """
    subfolder = str(self.subfolder).strip().strip("/")
    self.subfolder = ("/" + subfolder) if subfolder else ""

    mode = str(self.mode).strip().lower()
    if mode not in ("websockets", "webrtc"):
        logging.warning("Invalid mode value %r; using 'websockets'.", self.mode)
        mode = "websockets"
    self.mode = mode

    self.apply_webrtc_encoder_filter()

    if not self.enable_rate_control[0]:
        if (
            self._overridden.get("rate_control_mode")
            and self.rate_control_mode != "crf"
        ):
            logging.warning(
                "Ignoring rate_control_mode=%s: enable_rate_control is false, "
                "so the encoder runs CRF.",
                self.rate_control_mode,
            )
        self.rate_control_mode = "crf"
        rc_definition = next(
            (
                setting
                for setting in self._setting_definitions
                if setting["name"] == "rate_control_mode"
            ),
            None,
        )
        if rc_definition is not None:
            rc_definition["meta"]["allowed"] = ["crf"]
    else:
        self.resolve_rate_control_default()
    # Keys off the resolved mode whichever branch above produced it.
    self.resolve_paint_over_default()

    audio_enabled = self.audio_enabled[0]
    if not audio_enabled and self.microphone_enabled[0]:
        logging.warning(
            "Microphone support requires audio to be enabled. Disabling microphone support."
        )
        self.microphone_enabled = (False, self.microphone_enabled[1])

    mode = str(self.enable_clipboard).split("|")[0].strip().lower()
    if mode not in ("true", "false", "in", "out"):
        logging.warning(
            "Invalid enable_clipboard value %r; using 'true'.", self.enable_clipboard
        )
        mode = "true"
    self.enable_clipboard = mode

    if not self.turn_rest_username:
        self.turn_rest_username = "selkies"