-
-
Notifications
You must be signed in to change notification settings - Fork 30.5k
/
switch.py
146 lines (127 loc) · 4.67 KB
/
switch.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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
"""Switch platform for motionEye."""
from __future__ import annotations
from types import MappingProxyType
from typing import Any
from motioneye_client.client import MotionEyeClient
from motioneye_client.const import (
KEY_MOTION_DETECTION,
KEY_MOVIES,
KEY_STILL_IMAGES,
KEY_TEXT_OVERLAY,
KEY_UPLOAD_ENABLED,
KEY_VIDEO_STREAMING,
)
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import MotionEyeEntity, get_camera_from_cameras, listen_for_new_cameras
from .const import CONF_CLIENT, CONF_COORDINATOR, DOMAIN, TYPE_MOTIONEYE_SWITCH_BASE
MOTIONEYE_SWITCHES = [
SwitchEntityDescription(
key=KEY_MOTION_DETECTION,
translation_key="motion_detection",
entity_registry_enabled_default=True,
entity_category=EntityCategory.CONFIG,
),
SwitchEntityDescription(
key=KEY_TEXT_OVERLAY,
translation_key="text_overlay",
entity_registry_enabled_default=False,
entity_category=EntityCategory.CONFIG,
),
SwitchEntityDescription(
key=KEY_VIDEO_STREAMING,
translation_key="video_streaming",
entity_registry_enabled_default=False,
entity_category=EntityCategory.CONFIG,
),
SwitchEntityDescription(
key=KEY_STILL_IMAGES,
translation_key="still_images",
entity_registry_enabled_default=True,
entity_category=EntityCategory.CONFIG,
),
SwitchEntityDescription(
key=KEY_MOVIES,
translation_key="movies",
entity_registry_enabled_default=True,
entity_category=EntityCategory.CONFIG,
),
SwitchEntityDescription(
key=KEY_UPLOAD_ENABLED,
translation_key="upload_enabled",
entity_registry_enabled_default=False,
entity_category=EntityCategory.CONFIG,
),
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up motionEye from a config entry."""
entry_data = hass.data[DOMAIN][entry.entry_id]
@callback
def camera_add(camera: dict[str, Any]) -> None:
"""Add a new motionEye camera."""
async_add_entities(
[
MotionEyeSwitch(
entry.entry_id,
camera,
entry_data[CONF_CLIENT],
entry_data[CONF_COORDINATOR],
entry.options,
entity_description,
)
for entity_description in MOTIONEYE_SWITCHES
]
)
listen_for_new_cameras(hass, entry, camera_add)
class MotionEyeSwitch(MotionEyeEntity, SwitchEntity):
"""MotionEyeSwitch switch class."""
def __init__(
self,
config_entry_id: str,
camera: dict[str, Any],
client: MotionEyeClient,
coordinator: DataUpdateCoordinator,
options: MappingProxyType[str, str],
entity_description: SwitchEntityDescription,
) -> None:
"""Initialize the switch."""
super().__init__(
config_entry_id,
f"{TYPE_MOTIONEYE_SWITCH_BASE}_{entity_description.key}",
camera,
client,
coordinator,
options,
entity_description,
)
@property
def is_on(self) -> bool:
"""Return true if the switch is on."""
return bool(
self._camera and self._camera.get(self.entity_description.key, False)
)
async def _async_send_set_camera(self, value: bool) -> None:
"""Set a switch value."""
# Fetch the very latest camera config to reduce the risk of updating with a
# stale configuration.
camera = await self._client.async_get_camera(self._camera_id)
if camera:
camera[self.entity_description.key] = value
await self._client.async_set_camera(self._camera_id, camera)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
await self._async_send_set_camera(True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
await self._async_send_set_camera(False)
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._camera = get_camera_from_cameras(self._camera_id, self.coordinator.data)
super()._handle_coordinator_update()