Модели протокола¶

Pydantic-модели для валидации и конструирования JSON-структур протокола Sber Smart Home.

Pydantic models for Sber Smart Home MQTT protocol.

Provides strict typed schemas for Sber protocol payloads (device config, states, commands) and helper functions for constructing protocol values.

All models use extra="forbid" to reject unexpected fields — this catches protocol violations like the TV allowed_values bug (extra keys caused Sber cloud to silently reject devices).

These models serve as: - Executable specification of the Sber C2C JSON protocol - Pre-publish validation layer (invalid devices excluded from payload) - Type-safe constructors for Sber state values

Source of truth: https://developers.sber.ru/docs/ru/smarthome/c2c/

CATEGORY_REQUIRED_FEATURES `module-attribute` ¶

CATEGORY_REQUIRED_FEATURES = {cat: (get(cat, obligatory)) for cat, obligatory in (items())}

Required features per Sber category.

Derived from :data:CATEGORY_OBLIGATORY_FEATURES with the _CATEGORY_OBLIGATORY_OVERRIDES applied. Kept as the public name for backward compatibility — callers should continue using this constant.

SberColourValue ¶

Bases: BaseModel

HSV colour value per Sber spec.

Ranges

h: 0-360 (hue degrees) s: 0-1000 (saturation, 0.1% steps) v: 100-1000 (value/brightness, min 100 per Sber spec VR-004)

SberValue ¶

Bases: BaseModel

A typed value in Sber state or command payload.

Sber protocol uses tagged unions: the type field selects which *_value field carries the actual data.

Per Sber C2C spec: - integer_value is always a string (e.g. "220", not 220) - colour_value is an HSV object with h/s/v integer fields

SberState ¶

Bases: BaseModel

Single state key-value pair in the Sber protocol.

check_feature_type_matches ¶

check_feature_type_matches()

Reject states where the value type doesn't match the Sber spec for the key.

Protects against silent Sber rejections like the PIR-as-BOOL bug: Sber expects pir as {"type": "ENUM", "enum_value": "pir"} but a buggy device class could emit {"type": "BOOL", "bool_value": true} — Sber would accept the payload and quietly drop the device.

Uses :data:FEATURE_TYPES (auto-generated from Sber docs). Unknown keys are allowed — not all features are in the generated catalog (typos / undocumented features are handled by compliance tests).

Source code in custom_components/sber_mqtt_bridge/sber_models.py

@model_validator(mode="after")
def check_feature_type_matches(self) -> SberState:
    """Reject states where the value type doesn't match the Sber spec for the key.

    Protects against silent Sber rejections like the PIR-as-BOOL bug:
    Sber expects ``pir`` as ``{"type": "ENUM", "enum_value": "pir"}``
    but a buggy device class could emit ``{"type": "BOOL", "bool_value": true}``
    — Sber would accept the payload and quietly drop the device.

    Uses :data:`FEATURE_TYPES` (auto-generated from Sber docs).  Unknown
    keys are allowed — not all features are in the generated catalog
    (typos / undocumented features are handled by compliance tests).
    """
    expected = FEATURE_TYPES.get(self.key)
    if expected is not None and self.value.type != expected:
        raise ValueError(f"Feature {self.key!r} must have type {expected!r}, got {self.value.type!r}")
    return self

SberAllowedIntegerValues ¶

Bases: BaseModel

INTEGER allowed values with min/max/step as strings.

SberAllowedFloatValues ¶

Bases: BaseModel

FLOAT allowed values with numeric min/max.

SberAllowedEnumValues ¶

Bases: BaseModel

ENUM allowed values with list of valid strings.

SberAllowedValue ¶

Bases: BaseModel

Single allowed_values entry for a feature.

Type discriminator selects which *_values field is present. COLOUR type has no additional constraints — just {"type": "COLOUR"}.

SberDependencyCondition ¶

Bases: BaseModel

Single condition value in a dependency declaration.

SberDependency ¶

Bases: BaseModel

Feature dependency: feature X is available only when key Y has given values.

SberDeviceModel ¶

Bases: BaseModel

Device model descriptor within a Sber device config.

Per Sber spec, allowed_values should only contain keys that are in features and need non-default ranges. Extra keys cause silent device rejection.

must_have_online `classmethod` ¶

must_have_online(v)

All Sber devices must include 'online' feature (VR-010).

Source code in custom_components/sber_mqtt_bridge/sber_models.py

@field_validator("features")
@classmethod
def must_have_online(cls, v: list[str]) -> list[str]:
    """All Sber devices must include 'online' feature (VR-010)."""
    if "online" not in v:
        raise ValueError("'online' must be in features (VR-010)")
    return v

allowed_values_keys_must_be_known `classmethod` ¶

allowed_values_keys_must_be_known(v, info)

allowed_values keys must be subset of features (TV bug prevention).

Source code in custom_components/sber_mqtt_bridge/sber_models.py

@field_validator("allowed_values")
@classmethod
def allowed_values_keys_must_be_known(
    cls, v: dict[str, SberAllowedValue] | None, info: Any
) -> dict[str, SberAllowedValue] | None:
    """allowed_values keys must be subset of features (TV bug prevention)."""
    if v is None:
        return v
    features = info.data.get("features", [])
    extra_keys = set(v.keys()) - set(features)
    if extra_keys:
        raise ValueError(f"allowed_values contains keys not in features: {extra_keys}")
    return v

SberDevice ¶

Bases: BaseModel

Full device descriptor for Sber config publish.

Per Sber spec (VR-001), model_id and model are mutually exclusive. This integration always uses inline model; model_id is not emitted.

partner_meta_max_size `classmethod` ¶

partner_meta_max_size(v)

partner_meta JSON must be under 1024 chars (VR-003).

Source code in custom_components/sber_mqtt_bridge/sber_models.py

@field_validator("partner_meta")
@classmethod
def partner_meta_max_size(cls, v: dict | None) -> dict | None:
    """partner_meta JSON must be under 1024 chars (VR-003)."""
    if v is not None and len(json.dumps(v)) > 1024:
        raise ValueError("partner_meta JSON exceeds 1024 chars (VR-003)")
    return v

SberDeviceState ¶

Bases: BaseModel

Current state of a single device.

SberConfigPayload ¶

Bases: BaseModel

Config publish payload (up/config topic).

SberStatusPayload ¶

Bases: BaseModel

Status publish payload (up/status topic).

SberCommandPayload ¶

Bases: BaseModel

Incoming command payload from Sber cloud (down/commands topic).

make_bool_value ¶

make_bool_value(value)

Create a Sber BOOL value dict.

Parameters:

Name	Type	Description	Default
`value`	`bool`	Boolean value.	required

Returns:

Type	Description
`dict[str, Any]`	Dict ready for inclusion in a Sber state payload.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def make_bool_value(value: bool) -> dict[str, Any]:
    """Create a Sber BOOL value dict.

    Args:
        value: Boolean value.

    Returns:
        Dict ready for inclusion in a Sber state payload.
    """
    return {"type": "BOOL", "bool_value": value}

make_integer_value ¶

make_integer_value(value)

Create a Sber INTEGER value dict.

Per Sber C2C specification, integer_value is serialized as a string.

Parameters:

Name	Type	Description	Default
`value`	`int`	Integer value.	required

Returns:

Type	Description
`dict[str, Any]`	Dict ready for inclusion in a Sber state payload.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def make_integer_value(value: int) -> dict[str, Any]:
    """Create a Sber INTEGER value dict.

    Per Sber C2C specification, ``integer_value`` is serialized as a string.

    Args:
        value: Integer value.

    Returns:
        Dict ready for inclusion in a Sber state payload.
    """
    return {"type": "INTEGER", "integer_value": str(value)}

make_enum_value ¶

make_enum_value(value)

Create a Sber ENUM value dict.

Parameters:

Name	Type	Description	Default
`value`	`str`	Enum string value.	required

Returns:

Type	Description
`dict[str, Any]`	Dict ready for inclusion in a Sber state payload.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def make_enum_value(value: str) -> dict[str, Any]:
    """Create a Sber ENUM value dict.

    Args:
        value: Enum string value.

    Returns:
        Dict ready for inclusion in a Sber state payload.
    """
    return {"type": "ENUM", "enum_value": value}

make_colour_value ¶

make_colour_value(h, s, v)

Create a Sber COLOUR value dict.

Parameters:

Name	Type	Description	Default
`h`	`int`	Hue component (0-360).	required
`s`	`int`	Saturation component (0-1000).	required
`v`	`int`	Value/brightness component (100-1000).	required

Returns:

Type	Description
`dict[str, Any]`	Dict ready for inclusion in a Sber state payload.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def make_colour_value(h: int, s: int, v: int) -> dict[str, Any]:
    """Create a Sber COLOUR value dict.

    Args:
        h: Hue component (0-360).
        s: Saturation component (0-1000).
        v: Value/brightness component (100-1000).

    Returns:
        Dict ready for inclusion in a Sber state payload.
    """
    return {"type": "COLOUR", "colour_value": {"h": h, "s": s, "v": v}}

make_state ¶

make_state(key, value)

Create a Sber state entry dict.

Parameters:

Name	Type	Description	Default
`key`	`str`	State key name.	required
`value`	`dict[str, Any]`	Typed value dict (from `make_*_value` helpers).	required

Returns:

Type	Description
`dict[str, Any]`	Dict with `key` and `value` keys.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def make_state(key: str, value: dict[str, Any]) -> dict[str, Any]:
    """Create a Sber state entry dict.

    Args:
        key: State key name.
        value: Typed value dict (from ``make_*_value`` helpers).

    Returns:
        Dict with ``key`` and ``value`` keys.
    """
    return {"key": key, "value": value}

validate_device ¶

validate_device(device_data)

Validate a single device dict against SberDevice schema.

Parameters:

Name	Type	Description	Default
`device_data`	`dict[str, Any]`	Raw device dict from `to_sber_state()`.	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple `(valid, error_message)`. `error_message` is empty on success.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def validate_device(device_data: dict[str, Any]) -> tuple[bool, str]:
    """Validate a single device dict against SberDevice schema.

    Args:
        device_data: Raw device dict from ``to_sber_state()``.

    Returns:
        Tuple ``(valid, error_message)``.  ``error_message`` is empty on success.
    """
    try:
        SberDevice.model_validate(device_data)
    except (ValueError, TypeError) as exc:
        return False, str(exc)[:500]
    return True, ""

validate_config_payload ¶

validate_config_payload(data)

Validate a config payload dict against the SberConfigPayload schema.

This is an optional validation step — failures are logged as warnings but do not prevent publishing (the raw dict is still valid JSON).

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Raw dict to validate.	required

Returns:

Type	Description
`bool`	True if validation passed, False otherwise.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def validate_config_payload(data: dict[str, Any]) -> bool:
    """Validate a config payload dict against the SberConfigPayload schema.

    This is an optional validation step — failures are logged as warnings
    but do not prevent publishing (the raw dict is still valid JSON).

    Args:
        data: Raw dict to validate.

    Returns:
        True if validation passed, False otherwise.
    """
    try:
        SberConfigPayload.model_validate(data)
    except (ValueError, TypeError):
        _LOGGER.warning("Config payload validation failed", exc_info=True)
        return False
    return True

validate_status_payload ¶

validate_status_payload(data)

Validate a status payload dict against the SberStatusPayload schema.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Raw dict to validate.	required

Returns:

Type	Description
`bool`	True if validation passed, False otherwise.

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def validate_status_payload(data: dict[str, Any]) -> bool:
    """Validate a status payload dict against the SberStatusPayload schema.

    Args:
        data: Raw dict to validate.

    Returns:
        True if validation passed, False otherwise.
    """
    try:
        SberStatusPayload.model_validate(data)
    except (ValueError, TypeError):
        _LOGGER.warning("Status payload validation failed", exc_info=True)
        return False
    return True

missing_obligatory_features ¶

missing_obligatory_features(category, features)

Return obligatory features missing from the emitted set.

Uses :data:CATEGORY_OBLIGATORY_FEATURES (auto-generated from Sber docs ✔︎ markers). Missing obligatory features are a likely cause of silent device rejection by Sber cloud.

Parameters:

Name	Type	Description	Default
`category`	`str`	Sber category slug.	required
`features`	`set[str]`	Features our device would emit.	required

Returns:

Type	Description
`set[str]`	Set of obligatory features absent from `features`. Empty set
`set[str]`	for unknown categories (fail-open — we don't block unknown cats).

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def missing_obligatory_features(category: str, features: set[str]) -> set[str]:
    """Return obligatory features missing from the emitted set.

    Uses :data:`CATEGORY_OBLIGATORY_FEATURES` (auto-generated from Sber
    docs ``✔︎`` markers).  Missing obligatory features are a likely
    cause of silent device rejection by Sber cloud.

    Args:
        category: Sber category slug.
        features: Features our device would emit.

    Returns:
        Set of obligatory features absent from ``features``.  Empty set
        for unknown categories (fail-open — we don't block unknown cats).
    """
    obligatory = _effective_obligatory_features(category)
    if obligatory is None:
        return set()
    return obligatory - set(features)

unknown_features_for_category ¶

unknown_features_for_category(category, features)

Return features not present in Sber's reference model for the category.

Uses :data:CATEGORY_REFERENCE_FEATURES (auto-generated). Unknown categories (not in our registry) return empty set — validation falls back to general schema rules.

Parameters:

Name	Type	Description	Default
`category`	`str`	Sber category slug.	required
`features`	`set[str]`	Features our device would emit.	required

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def unknown_features_for_category(category: str, features: set[str]) -> set[str]:
    """Return features not present in Sber's reference model for the category.

    Uses :data:`CATEGORY_REFERENCE_FEATURES` (auto-generated).  Unknown
    categories (not in our registry) return empty set — validation falls
    back to general schema rules.

    Args:
        category: Sber category slug.
        features: Features our device would emit.
    """
    reference = CATEGORY_REFERENCE_FEATURES.get(category)
    if reference is None:
        return set()
    return set(features) - reference

validate_category_compliance ¶

validate_category_compliance(device)

Check a device descriptor for Sber category-specific violations.

Returns a list of human-readable violation messages (empty = compliant). Does NOT raise — callers decide how to handle violations.

Parameters:

Name	Type	Description	Default
`device`	`dict[str, Any]`	Raw device dict (already passed SberDevice schema validation).	required

Source code in custom_components/sber_mqtt_bridge/sber_models.py

def validate_category_compliance(device: dict[str, Any]) -> list[str]:
    """Check a device descriptor for Sber category-specific violations.

    Returns a list of human-readable violation messages (empty = compliant).
    Does NOT raise — callers decide how to handle violations.

    Args:
        device: Raw device dict (already passed SberDevice schema validation).
    """
    violations: list[str] = []
    model = device.get("model", {})
    category = model.get("category", "")
    features = set(model.get("features", []))

    # VR-010..VR-016: required features per category
    required = CATEGORY_REQUIRED_FEATURES.get(category)
    if required is not None:
        missing = required - features
        if missing:
            violations.append(f"Missing required features for {category}: {missing}")

    # TV bug prevention: allowed_values keys must be subset of features
    allowed_values = model.get("allowed_values") or {}
    extra_av = set(allowed_values.keys()) - features
    if extra_av:
        violations.append(f"allowed_values contains keys not in features: {extra_av}")

    return violations

Модели протокола¶

CATEGORY_REQUIRED_FEATURES module-attribute ¶

SberColourValue ¶

SberValue ¶

SberState ¶

check_feature_type_matches ¶

SberAllowedIntegerValues ¶

SberAllowedFloatValues ¶

SberAllowedEnumValues ¶

SberAllowedValue ¶

SberDependencyCondition ¶

SberDependency ¶

SberDeviceModel ¶

must_have_online classmethod ¶

allowed_values_keys_must_be_known classmethod ¶

SberDevice ¶

partner_meta_max_size classmethod ¶

SberDeviceState ¶

SberConfigPayload ¶

SberStatusPayload ¶

SberCommandPayload ¶

make_bool_value ¶

make_integer_value ¶

make_enum_value ¶

make_colour_value ¶

make_state ¶

validate_device ¶

validate_config_payload ¶

validate_status_payload ¶

missing_obligatory_features ¶

unknown_features_for_category ¶

validate_category_compliance ¶

CATEGORY_REQUIRED_FEATURES `module-attribute` ¶

must_have_online `classmethod` ¶

allowed_values_keys_must_be_known `classmethod` ¶

partner_meta_max_size `classmethod` ¶