Перейти к содержанию

Медиа

TVEntity

Телевизор: управление питанием, громкостью, каналами, источником.

Sber TV entity -- maps HA media_player entities to Sber tv category.

Supports on/off, volume, mute, source selection, channel switching, navigation direction, and custom key commands.

TV_CATEGORY module-attribute 

TV_CATEGORY = 'tv'

Sber device category for TV entities.

TvEntity 

TvEntity(entity_data)

Bases: BaseEntity

Sber TV entity for television and media player devices.

Maps HA media_player entities to the Sber 'tv' category with support for: - On/off control - Volume level (Sber 0-100 integer, HA 0.0-1.0 float) - Mute toggle - Source (input) selection - Channel switching (+/-) - Navigation direction (up/down/left/right/ok)

Initialize TV entity.

Parameters:

Name Type Description Default
entity_data dict

HA entity registry dict containing entity metadata.

 required
Source code in custom_components/sber_mqtt_bridge/devices/tv.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(self, entity_data: dict) -> None:
    """Initialize TV entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(TV_CATEGORY, entity_data)
    self.current_state: bool = False
    self._volume: int = 0
    self._is_muted: bool = False
    self._source: str | None = None
    self._source_list: list[str] = []
    self._media_content_id: str | None = None

fill_by_ha_state 

fill_by_ha_state(ha_state)

Parse HA state and update TV attributes.

Parameters:

Name Type Description Default
ha_state dict

HA state dict with 'state' and 'attributes' keys.

 required
Source code in custom_components/sber_mqtt_bridge/devices/tv.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update TV attributes.

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    self.current_state = ha_state.get("state") not in ("off", "standby", "unavailable", "unknown")
    attrs = ha_state.get("attributes", {})
    vol = self._safe_float(attrs.get("volume_level"))
    self._volume = int(vol * 100) if vol is not None else 0
    self._is_muted = bool(attrs.get("is_volume_muted", False))
    self._source = attrs.get("source")
    self._source_list = attrs.get("source_list") or []
    self._media_content_id = attrs.get("media_content_id")

create_features_list 

create_features_list()

Return Sber feature list for TV capabilities.

Returns:

Type Description
list[str]

List of Sber feature strings supported by this entity.
Source code in custom_components/sber_mqtt_bridge/devices/tv.py 
63
64
65
66
67
68
69
70
71
72
73
def create_features_list(self) -> list[str]:
    """Return Sber feature list for TV capabilities.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    features = [*super().create_features_list(), "on_off", "volume_int", "mute"]
    if self._source_list:
        features.append("source")
    features.extend(["channel", "channel_int", "direction"])
    return features

create_allowed_values_list 

create_allowed_values_list()

Build allowed values map for volume and source features.

Returns:

Type Description
dict[str, dict]

Dict mapping feature key to its allowed values descriptor.
Source code in custom_components/sber_mqtt_bridge/devices/tv.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def create_allowed_values_list(self) -> dict[str, dict]:
    """Build allowed values map for volume and source features.

    Returns:
        Dict mapping feature key to its allowed values descriptor.
    """
    allowed: dict[str, dict] = {
        "volume_int": {
            "type": "INTEGER",
            "integer_values": {"min": "0", "max": "100", "step": "1"},
        },
        "channel": {
            "type": "ENUM",
            "enum_values": {"values": ["+", "-"]},
        },
        "channel_int": {
            "type": "INTEGER",
            "integer_values": {"min": "1", "max": "999", "step": "1"},
        },
        "direction": {
            "type": "ENUM",
            "enum_values": {"values": ["up", "down", "left", "right", "ok"]},
        },
    }
    if self._source_list:
        allowed["source"] = {
            "type": "ENUM",
            "enum_values": {"values": self._source_list},
        }
    return allowed

to_sber_current_state 

to_sber_current_state()

Build Sber current state payload with TV attributes.

Returns:

Type Description
dict[str, dict]

Dict mapping entity_id to its Sber state representation.
Source code in custom_components/sber_mqtt_bridge/devices/tv.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with TV attributes.

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    states = [
        make_state(SberFeature.ONLINE, make_bool_value(self._is_online)),
        make_state(SberFeature.ON_OFF, make_bool_value(self.current_state)),
        make_state(SberFeature.VOLUME_INT, make_integer_value(self._volume)),
        make_state(SberFeature.MUTE, make_bool_value(self._is_muted)),
    ]
    if self._source:
        states.append(make_state(SberFeature.SOURCE, make_enum_value(self._source)))
    return {self.entity_id: {"states": states}}

process_cmd 

process_cmd(cmd_data)

Process Sber TV commands and produce HA service calls.

Handles the following Sber keys: - on_off: media_player.turn_on / media_player.turn_off - volume_int: media_player.volume_set (Sber 0-100 → HA 0.0-1.0) - mute: media_player.volume_mute - source: media_player.select_source

Parameters:

Name Type Description Default
cmd_data dict

Sber command dict with 'states' list.

 required

Returns:

Type Description
list[dict]

List of HA service call dicts to execute.
Source code in custom_components/sber_mqtt_bridge/devices/tv.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber TV commands and produce HA service calls.

    Handles the following Sber keys:
    - ``on_off``: media_player.turn_on / media_player.turn_off
    - ``volume_int``: media_player.volume_set (Sber 0-100 → HA 0.0-1.0)
    - ``mute``: media_player.volume_mute
    - ``source``: media_player.select_source

    Args:
        cmd_data: Sber command dict with 'states' list.

    Returns:
        List of HA service call dicts to execute.
    """
    results: list[dict] = []
    for item in cmd_data.get("states", []):
        key = item.get("key")
        value = item.get("value", {})

        if key == "on_off" and value.get("type") == "BOOL":
            on = value.get("bool_value", False)
            results.append(self._build_on_off_service_call(self.entity_id, "media_player", on))

        elif key == "volume_int" and value.get("type") == "INTEGER":
            raw = value.get("integer_value")
            if raw is None:
                continue
            vol = self._safe_int(raw)
            if vol is None:
                continue
            ha_volume = vol / 100.0
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "media_player",
                        "service": "volume_set",
                        "service_data": {"volume_level": ha_volume},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )

        elif key == "mute" and value.get("type") == "BOOL":
            muted = value.get("bool_value", False)
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "media_player",
                        "service": "volume_mute",
                        "service_data": {"is_volume_muted": muted},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )

        elif key == "source" and value.get("type") == "ENUM":
            source = value.get("enum_value")
            if not source:
                continue
            results.append(
                {
                    "url": {
                        "type": "call_service",
                        "domain": "media_player",
                        "service": "select_source",
                        "service_data": {"source": source},
                        "target": {"entity_id": self.entity_id},
                    }
                }
            )

        elif key == "channel_int" and value.get("type") == "INTEGER":
            raw_ch = value.get("integer_value")
            if raw_ch is None:
                continue
            ch = self._safe_int(raw_ch)
            if ch is not None:
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "media_player",
                            "service": "play_media",
                            "service_data": {
                                "media_content_type": "channel",
                                "media_content_id": str(ch),
                            },
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )

        elif key == "channel" and value.get("type") == "ENUM":
            direction = value.get("enum_value")
            if direction == "+":
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "media_player",
                            "service": "media_next_track",
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )
            elif direction == "-":
                results.append(
                    {
                        "url": {
                            "type": "call_service",
                            "domain": "media_player",
                            "service": "media_previous_track",
                            "target": {"entity_id": self.entity_id},
                        }
                    }
                )

        elif key == "direction" and value.get("type") == "ENUM":
            direction_cmd = value.get("enum_value")
            if not direction_cmd:
                continue
            # Map Sber direction to HA media_player commands
            # Sber docs: up, down, left, right, ok
            if direction_cmd == "up":
                results.append(
                    {"url": {"type": "call_service", "domain": "media_player", "service": "volume_up", "target": {"entity_id": self.entity_id}}}
                )
            elif direction_cmd == "down":
                results.append(
                    {"url": {"type": "call_service", "domain": "media_player", "service": "volume_down", "target": {"entity_id": self.entity_id}}}
                )
            elif direction_cmd == "left":
                results.append(
                    {"url": {"type": "call_service", "domain": "media_player", "service": "media_previous_track", "target": {"entity_id": self.entity_id}}}
                )
            elif direction_cmd == "right":
                results.append(
                    {"url": {"type": "call_service", "domain": "media_player", "service": "media_next_track", "target": {"entity_id": self.entity_id}}}
                )
            elif direction_cmd == "ok":
                results.append(
                    {"url": {"type": "call_service", "domain": "media_player", "service": "media_play_pause", "target": {"entity_id": self.entity_id}}}
                )

    return results

IntercomEntity

Домофон.

Sber Intercom entity -- minimal implementation for intercom devices.

Available only via type override (sber_category: intercom). Supports on/off control and read-only call/unlock features from HA attributes.

INTERCOM_CATEGORY module-attribute 

INTERCOM_CATEGORY = 'intercom'

Sber device category for intercom entities.

IntercomEntity 

IntercomEntity(entity_data)

Bases: OnOffEntity

Sber intercom entity for door intercom devices.

Maps to the Sber 'intercom' category. Available only via type override since there is no standard HA intercom domain.

Supports: - On/off control (inherited from OnOffEntity) - Read-only features from HA attributes: incoming_call, reject_call, unlock

Initialize intercom entity.

Parameters:

Name Type Description Default
entity_data dict

HA entity registry dict containing entity metadata.

 required
Source code in custom_components/sber_mqtt_bridge/devices/intercom.py 
32
33
34
35
36
37
38
39
40
41
def __init__(self, entity_data: dict) -> None:
    """Initialize intercom entity.

    Args:
        entity_data: HA entity registry dict containing entity metadata.
    """
    super().__init__(INTERCOM_CATEGORY, entity_data)
    self._incoming_call: bool = False
    self._reject_call: bool = False
    self._unlock: bool = False

fill_by_ha_state 

fill_by_ha_state(ha_state)

Parse HA state and update intercom attributes.

Parameters:

Name Type Description Default
ha_state dict

HA state dict with 'state' and 'attributes' keys.

 required
Source code in custom_components/sber_mqtt_bridge/devices/intercom.py 
43
44
45
46
47
48
49
50
51
52
53
def fill_by_ha_state(self, ha_state: dict) -> None:
    """Parse HA state and update intercom attributes.

    Args:
        ha_state: HA state dict with 'state' and 'attributes' keys.
    """
    super().fill_by_ha_state(ha_state)
    attrs = ha_state.get("attributes", {})
    self._incoming_call = bool(attrs.get("incoming_call", False))
    self._reject_call = bool(attrs.get("reject_call", False))
    self._unlock = bool(attrs.get("unlock", False))

create_features_list 

create_features_list()

Return Sber feature list for intercom capabilities.

Returns:

Type Description
list[str]

List of Sber feature strings supported by this entity.
Source code in custom_components/sber_mqtt_bridge/devices/intercom.py 
55
56
57
58
59
60
61
62
63
64
65
66
def create_features_list(self) -> list[str]:
    """Return Sber feature list for intercom capabilities.

    Returns:
        List of Sber feature strings supported by this entity.
    """
    return [
        *super().create_features_list(),
        "incoming_call",
        "reject_call",
        "unlock",
    ]

to_sber_current_state 

to_sber_current_state()

Build Sber current state payload with intercom attributes.

Returns:

Type Description
dict[str, dict]

Dict mapping entity_id to its Sber state representation.
Source code in custom_components/sber_mqtt_bridge/devices/intercom.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def to_sber_current_state(self) -> dict[str, dict]:
    """Build Sber current state payload with intercom attributes.

    Returns:
        Dict mapping entity_id to its Sber state representation.
    """
    base = super().to_sber_current_state()
    states = base[self.entity_id]["states"]
    states.extend(
        [
            make_state(SberFeature.INCOMING_CALL, make_bool_value(self._incoming_call)),
            make_state(SberFeature.REJECT_CALL, make_bool_value(self._reject_call)),
            make_state(SberFeature.UNLOCK, make_bool_value(self._unlock)),
        ]
    )
    return base

process_cmd 

process_cmd(cmd_data)

Process Sber intercom commands and produce HA service calls.

Handles on_off key for turn_on/turn_off. Other features (incoming_call, reject_call, unlock) are read-only.

Parameters:

Name Type Description Default
cmd_data dict

Sber command dict with 'states' list.

 required

Returns:

Type Description
list[dict]

List of HA service call dicts to execute.
Source code in custom_components/sber_mqtt_bridge/devices/intercom.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def process_cmd(self, cmd_data: dict) -> list[dict]:
    """Process Sber intercom commands and produce HA service calls.

    Handles ``on_off`` key for turn_on/turn_off. Other features
    (incoming_call, reject_call, unlock) are read-only.

    Args:
        cmd_data: Sber command dict with 'states' list.

    Returns:
        List of HA service call dicts to execute.
    """
    results: list[dict] = []
    domain = self.get_entity_domain()
    for item in cmd_data.get("states", []):
        key = item.get("key")
        value = item.get("value", {})
        if key == "on_off" and value.get("type") == "BOOL":
            on = value.get("bool_value", False)
            results.append(self._build_on_off_service_call(self.entity_id, domain, on))
    return results