Selkies

AppSettings

Parses and stores application settings from CLI arguments and environment variables, based on the centralized definition list.

Each setting becomes an instance attribute named after its definition: bool settings resolve to a (value, locked) tuple, range settings to a (min, max) tuple (with the initial value kept in the definition's meta["default_value"]), and other types to their parsed scalar/list value. was_provided reports whether an operator set a value explicitly, which drives conditional defaults and operator locks downstream. The class-level annotations type the settings static tools access by attribute; every other setting is attached dynamically by _process_and_set_attributes.

Attributes

attributedebugtuple[bool, bool]
attributegamepad_enabledtuple[bool, bool]
attributecommand_enabledtuple[bool, bool]
attributewebrtc_pacertuple[bool, bool]
attributeuinput_mouse_socketstr
attributeuinput_gamepadstr
attributepublish_input_devicestuple[bool, bool]
attributewebcam_enabledtuple[bool, bool]
attributewebcam_socket_pathstr
attributewebcam_widthint
attributewebcam_heightint
attributewebcam_pixel_formatstr
attributewebcam_devicestr
attributeenable_cursorstuple[bool, bool]
attributedebug_cursorstuple[bool, bool]
attributeenable_resizetuple[bool, bool]
attributeaudio_bitratestr
attributewaylandtuple[bool, bool]
attributecursor_sizeint
attributegpu_idstr
attributeaudio_enabledtuple[bool, bool]
attributeenable_collabtuple[bool, bool]
attributemaster_tokenstr
attributevideo_fullcolortuple[bool, bool]
attributevideo_10bittuple[bool, bool]
attributesubfolderstr
attributevideo_bitratetuple[float, float]
attributefile_transfer_limit_mbpsfloat
attribute_setting_definitions
= setting

The definition list, mutated in place when an override narrows a menu.

Functions

constructor__init__(setting) -> 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.

Source Code
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",
        epilog=help_epilog(),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    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()
paramsettingList[Dict[str, Any]]

Returns

None
func_fallback_env_vars(setting) -> 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).

Source Code
@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)
paramsettingDict[str, Any]

Returns

typing.Tuple[str, ...]
func_add_arguments(parser) -> 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.

Source Code
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 {}),
        )
paramparserargparse.ArgumentParser

Returns

None
func_process_and_set_attributes(args) -> 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.

Source Code
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
paramargsargparse.Namespace

Returns

None
funcwas_provided(name) -> 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.

Source Code
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))
paramnamestr

Returns

bool
funcencode_node_index() -> 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.

Source Code
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

Returns

typing.Optional[int]
funcencoder_backends() -> 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), 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), and ten_bit, whether each side takes a video_10bit session at 10 bits, by chroma ("420", "444"; None where the side is absent or unknown). None before resolve_encoder_backends ran or where the hardware side is unknown, so no consumer hides a choice on a guess.

Source Code
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), `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), and `ten_bit`, whether each side
    takes a `video_10bit` session at 10 bits, by chroma (`"420"`, `"444"`; None
    where the side is absent or unknown). 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

    hw_formats = getattr(self, "_hardware_formats", None)
    sw_formats = software_formats()

    def ten_bit(codec: str, backend: Optional[str],
                formats: Optional[Dict[str, List[str]]]) -> Optional[Dict[str, bool]]:
        if backend is None or formats is None:
            return None
        listed = formats.get(codec, [])
        return {chroma: f"{chroma}-10" in listed for chroma in ("420", "444")}

    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),
            },
            "ten_bit": {
                "hardware": ten_bit(codec, hardware.get(codec), hw_formats),
                "software": ten_bit(codec, software.get(codec), sw_formats),
            },
        }
        for codec in CODEC_LABELS
        if codec != "jpeg"
    }

Returns

typing.Optional[typing.Dict[str, typing.Dict[str, typing.Any]]]
funcencoder_fullcolor(encoder, use_cpu=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. A VA-API engine refuses a 4:4:4 it lacks rather than streaming 4:2:0, so that session is the software encoder's too (H.264 on every Intel and AMD device). None where the side that answers is unknown.

Source Code
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. A VA-API engine refuses a 4:4:4 it
    lacks rather than streaming 4:2:0, so that session is the software encoder's too
    (H.264 on every Intel and AMD device). None where the side that answers is unknown."""
    backends = self.encoder_backends()
    served = (backends or {}).get(codec_for_encoder(canonical_encoder(encoder)))
    if not served:
        return None
    carried = served["fullcolor"]
    if use_cpu or encoder in CPU_ONLY_ENCODERS or not served["hardware"]:
        return carried["software"]
    if carried["hardware"] is False and served["hardware"] == "vaapi" and carried["software"]:
        return True
    return carried["hardware"]
paramencoderstr
paramuse_cpubool
= False

Returns

typing.Optional[bool]
funcencoder_ten_bit(encoder, use_cpu=False, fullcolor=False) -> Optional[bool]

Whether a video_10bit session on this encoder streams 10 bits from this host.

A session runs on the encode node's engine where the codec has one and software is not forced or striped, and pixelflux hands it to the build's software encoder where the engine lacks the 4:4:4 a full-color session asks for, or the 10 bits this one does while the software encoder codes them. Each side answers at the chroma it would run: its 4:4:4 where full color is on and it carries that, else 4:2:0.

Source Code
def encoder_ten_bit(self, encoder: str, use_cpu: bool = False, fullcolor: bool = False) -> Optional[bool]:
    """Whether a `video_10bit` session on this encoder streams 10 bits from this host.

    A session runs on the encode node's engine where the codec has one and software
    is not forced or striped, and pixelflux hands it to the build's software encoder
    where the engine lacks the 4:4:4 a full-color session asks for, or the 10 bits
    this one does while the software encoder codes them. Each side answers at the
    chroma it would run: its 4:4:4 where full color is on and it carries that, else
    4:2:0.

    Args:
        encoder: The encoder wire value.
        use_cpu: Whether software encoding is forced.
        fullcolor: Whether the session asks for 4:4:4.

    Returns:
        Whether the stream is 10-bit; None where the side that would answer is unknown.
    """
    backends = self.encoder_backends()
    served = (backends or {}).get(codec_for_encoder(canonical_encoder(encoder)))
    if not served:
        return None

    def carried(side: str) -> Optional[bool]:
        table = served["ten_bit"][side]
        if table is None:
            return None
        return table["444" if fullcolor and served["fullcolor"][side] else "420"]

    if use_cpu or encoder in CPU_ONLY_ENCODERS or not served["hardware"]:
        return carried("software")
    if fullcolor and served["fullcolor"]["hardware"] is False and served["fullcolor"]["software"]:
        return carried("software")
    return carried("hardware") or carried("software") or carried("hardware")
paramencoderstr

The encoder wire value.

paramuse_cpubool
= False

Whether software encoding is forced.

paramfullcolorbool
= False

Whether the session asks for 4:4:4.

Returns

typing.Optional

Whether the stream is 10-bit; None where the side that would answer is unknown.

funcencoder_served(encoder) -> 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.

Source Code
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"))
paramencoderstr

Returns

bool
funcresolve_encoder_backends() -> 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.

Source Code
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 ""))
    self._hardware_formats = {} if node is None else hardware_formats(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()

Returns

None
funcapply_webrtc_encoder_filter() -> 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.

Source Code
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

Returns

None
funcresolve_paint_over_default() -> None

Default paint-over off while Turbo drives a video encoder and on otherwise, JPEG included: the dashboards' default (USE_PAINT_OVER_QUALITY_SPEC), so a client without one gets it as well. A constant rate cleans up a still screen to the paint-over quality either way, so the default matters to CRF sessions and JPEG.

An operator's use_paint_over_quality wins; a client's choice lives in its display's state. Called again on a transport switch, which can change the encoder.

Source Code
def resolve_paint_over_default(self) -> None:
    """Default paint-over off while Turbo drives a video encoder and on otherwise, JPEG
    included: the dashboards' default (`USE_PAINT_OVER_QUALITY_SPEC`), so a client without
    one gets it as well. A constant rate cleans up a still screen to the paint-over quality
    either way, so the default matters to CRF sessions and JPEG.

    An operator's use_paint_over_quality wins; a client's choice lives in its display's
    state. Called again on a transport switch, which can change the encoder.
    """
    if self.was_provided("use_paint_over_quality"):
        return
    self.use_paint_over_quality = (
        canonical_encoder(self.encoder) == "jpeg" or not self.video_streaming_mode[0],
        self.use_paint_over_quality[1],
    )

Returns

None
funcresolve_cpu_encode_threading_default() -> None

Default the software encode threads off for a VP8 session and on for any other (effective_cpu_encode_threading), so the payload a page first takes already says what the session encoder runs with.

An operator's cpu_encode_threading wins; a client's choice lives in its display's state. Called again on a transport switch, which can change the encoder.

Source Code
def resolve_cpu_encode_threading_default(self) -> None:
    """Default the software encode threads off for a VP8 session and on for any other
    (`effective_cpu_encode_threading`), so the payload a page first takes already says what
    the session encoder runs with.

    An operator's cpu_encode_threading wins; a client's choice lives in its display's
    state. Called again on a transport switch, which can change the encoder.
    """
    if self.was_provided("cpu_encode_threading"):
        return
    self.cpu_encode_threading = (
        effective_cpu_encode_threading(self.encoder, None, True, False),
        self.cpu_encode_threading[1],
    )

Returns

None
func_post_process_settings() -> 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; the "cbr" default would leave the dashboards showing a bitrate slider the encoder ignores and hiding the CRF slider in force. Paint-over defaults off where Turbo drives a video encoder (resolve_paint_over_default). 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.

Source Code
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; the "cbr" default would leave the
    dashboards showing a bitrate slider the encoder ignores and hiding the
    CRF slider in force. Paint-over defaults off where Turbo drives a video
    encoder (`resolve_paint_over_default`). 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"]
    self.resolve_paint_over_default()
    self.resolve_cpu_encode_threading_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()
    # The bool spellings every other switch takes, so "0" turns the clipboard off.
    mode = {"1": "true", "0": "false"}.get(mode, mode)
    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,::"

Returns

None