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()
    self._process_and_set_attributes(args)
    configure_logging(bool(self.debug[0]))
    for token in unknown:
        if token.startswith("-"):
            logger.warning(
                "Ignoring unrecognized argument %s", token.split("=", 1)[0]
            )
    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; a bool flag given bare reads as
    "true". 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})",
            **({"nargs": "?", "const": "true"} if setting["type"] == "bool" else {}),
        )

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:
                                logger.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:
                    logger.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:
                            logger.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:
            logger.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:
        logger.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
            logger.info("Manual width not set or invalid, defaulting to 1024.")
        if processed.get("manual_height", 0) <= 0:
            processed["manual_height"] = 768
            logger.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 encode_node_index(self) -> Optional[int]:
    """The DRI render-node index hardware encoders open, resolved as the
    capture settings resolve it: `encode_dri` names a node, else `gpu_id`
    picks one, else `AUTO_ENCODE_NODE` leaves the pick to pixelflux's
    `auto_gpu` selection. None where no session encodes on hardware:
    `gpu_id` -1, an unusable `encode_dri`, or software encoding locked on."""
    from .display_utils import parse_dri_node_to_index, parse_gpu_id
    if self.use_cpu[0] and self.use_cpu[1]:
        return None
    node = str(getattr(self, "encode_dri", "") or "")
    if node:
        index = parse_dri_node_to_index(node)
        return None if index < 0 else index
    gid = parse_gpu_id(self.gpu_id)
    if gid is None:
        return AUTO_ENCODE_NODE
    return None if gid < 0 else gid

def encoder_backends(self) -> Optional[Dict[str, Dict[str, Any]]]:
    """The backends that serve each video codec on this host, by codec
    name: `hardware` (the backend pixelflux named, or None) from the startup probe,
    `software` (the pixelflux build's library or None), and `fullcolor`, whether
    each of those sides takes a `video_fullcolor` session as 4:4:4 (None where the
    side is absent or the build does not say). None before
    `resolve_encoder_backends` ran or where the hardware side is unknown,
    so no consumer hides a choice on a guess."""
    hardware = getattr(self, "_hardware_encoders", None)
    if hardware is None:
        return None
    software = software_encoders()
    hw_fullcolor = getattr(self, "_hardware_fullcolor", None)
    sw_fullcolor = software_fullcolor()

    def carries(codec: str, backend: Optional[str], table: Optional[List[str]]) -> Optional[bool]:
        return None if backend is None or table is None else codec in table

    return {
        codec: {
            "hardware": hardware.get(codec),
            "software": software.get(codec),
            "fullcolor": {
                "hardware": carries(codec, hardware.get(codec), hw_fullcolor),
                "software": carries(codec, software.get(codec), sw_fullcolor),
            },
        }
        for codec in CODEC_LABELS
        if codec != "jpeg"
    }

def encoder_fullcolor(self, encoder: str, use_cpu: bool = False) -> Optional[bool]:
    """Whether a `video_fullcolor` session on this encoder streams 4:4:4 from this host:
    by the encode node's engine where the codec has one and software is not forced or
    striped, else by the build's software encoder. None where that side is unknown."""
    backends = self.encoder_backends()
    served = (backends or {}).get(codec_for_encoder(canonical_encoder(encoder)))
    if not served:
        return None
    side = "software" if use_cpu or encoder in CPU_ONLY_ENCODERS or not served["hardware"] else "hardware"
    return served["fullcolor"][side]

def encoder_served(self, encoder: str) -> bool:
    """Whether a session on this encoder comes up on the codec it names
    rather than demoting: the CPU-only encoders always, a full-frame one
    where its codec has a software encoder in the build or a hardware one
    on the encode node. True for every encoder while the hardware side is
    unknown."""
    backends = self.encoder_backends()
    if backends is None:
        return True
    encoder = canonical_encoder(encoder)
    if encoder in CPU_ONLY_ENCODERS:
        return True
    served = backends.get(codec_for_encoder(encoder), {})
    return bool(served.get("software") or served.get("hardware"))

def resolve_encoder_backends(self) -> None:
    """Learn once, at startup, which encoders this host serves, and narrow
    the encoder menu to them.

    The hardware side comes from pixelflux's per-node probe on the encode
    node the capture settings resolve; a host that never encodes on
    hardware (`gpu_id` -1, software encoding locked on) has none. The
    operator's menu keeps only the encoders that come up on their codec,
    so a client is never offered one the selection ladder would demote,
    and the transport filter re-derives its view from the narrowed menu.
    An operator's pick that is not served falls back to H.264 with a
    warning, the codec every host serves; a menu narrowed to nothing
    reverts to every served encoder rather than offering none.
    """
    node = self.encode_node_index()
    self._hardware_encoders = ({} if node is None
                               else hardware_encoders(node, str(self.auto_gpu or "")))
    self._hardware_fullcolor = [] if node is None else hardware_fullcolor(node, str(self.auto_gpu or ""))
    if self._hardware_encoders is None:
        return
    enc_definition = next(
        (s for s in self._setting_definitions if s["name"] == "encoder"),
        None,
    )
    if enc_definition is None:
        return
    operator_allowed = list(getattr(self, "_operator_encoder_allowed", enc_definition["meta"]["allowed"]))
    served = [item for item in operator_allowed if self.encoder_served(item)]
    dropped = [item for item in operator_allowed if item not in served]
    if not served:
        served = [item for item in ENCODER_CODECS if self.encoder_served(item)]
        logger.warning(
            "No encoder of the configured menu (%s) is served on this host; offering %s.",
            ", ".join(operator_allowed),
            ", ".join(served),
        )
    elif dropped:
        logger.info("Encoders not served on this host are left off the menu: %s", ", ".join(dropped))
    self._operator_encoder_allowed = served
    value = getattr(self, "_operator_encoder_value", self.encoder)
    if value not in served:
        backends = self.encoder_backends()
        fallback = min(served, key=lambda item: encoder_rung(item, backends))
        logger.warning("Encoder %r is not served on this host; using %r.", value, fallback)
        self._operator_encoder_value = fallback
        if self.encoder == value:
            self.encoder = fallback
    enc_definition["meta"]["allowed"] = list(served)
    self.apply_webrtc_encoder_filter()

def on_software_video_path(self) -> bool:
    """Whether the server's own defaults put a session on the software
    video path: the striped encoder, or a full-frame encoder 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_video_path(self.encoder, forced)

def software_encoder_in_use(self) -> Optional[str]:
    """The software encoder a session on the software path encodes with,
    by the pixelflux build's table; None off that path or without one."""
    if not self.on_software_video_path():
        return None
    return software_encoders().get(codec_for_encoder(self.encoder))

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_encoders`.

    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.software_encoder_in_use() == "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 = [item for item in WEBRTC_ENCODER_CHOICES if self.encoder_served(item)]
    enc_definition["meta"]["allowed"] = allowed
    if self.encoder not in allowed:
        fallback = min(allowed, key=lambda item: encoder_rung(item, self.encoder_backends()))
        if self.was_provided("encoder"):
            logger.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.
    A public listener is the both-family wildcard address, so the server
    binds from `addr` alone.
    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"):
        logger.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"
        ):
            logger.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]:
        logger.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"):
        logger.warning(
            "Invalid enable_clipboard value %r; using 'true'.", self.enable_clipboard
        )
        mode = "true"
    self.enable_clipboard = mode

    for name in ("microphone_on_start", "webcam_on_start"):
        mode = str(getattr(self, name)).split("|")[0].strip().lower()
        # The bool spellings these settings once took are still read.
        mode = {"1": "true", "0": "false"}.get(mode, mode)
        if mode not in ("true", "false", "demand"):
            logger.warning("Invalid %s value %r; using 'false'.", name, getattr(self, name))
            mode = "false"
        setattr(self, name, mode)

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

    if self.public[0]:
        self.addr = "0.0.0.0,::"