Path:
src/midi_bulb.py
Lines:
187
Non-empty lines:
160
Non-empty lines covered with requirements:
21 / 160 (13.1%)
Functions:
15
Functions covered by requirements:
1 / 15 (6.7%)
1
"""2
3
"""4
from contextlib import contextmanager, suppress
5
from typing import Any, Dict, Generator, List, Optional, Sequence, Tuple
6
import consts as C
7
import yeelight as Y
8
# import dev.yeelight_dummy as Y # For development purposes, replace with actual yeelight import in production9
import logging
10
import yaml
11
12
13
logger = logging.getLogger(__name__)
14
CAPABILITIES = "capabilities"
15
ID = "id"
16
IP = "ip"
17
__discovered: Optional[Dict] = None
18
19
20
def get_discovered() -> List[Dict]:
21
"""
22
Returns the discovered list of dictionaries describing bulbs.23
Else, queries the network for available bulbs.24
25
:return: List of Dicts or None26
27
@relation(YMD-SYS-3, scope=function)28
"""29
global __discovered
30
if __discovered is None:
31
try:
32
logger.info("Discovering bulbs...")
33
__discovered = Y.discover_bulbs()
34
if len(__discovered) < 1:
35
raise ValueError("No bulbs discovered.")
36
for bulb in __discovered:
37
logger.info(
38
f"Discovered bulb: {bulb[CAPABILITIES][ID]} at {bulb[IP]}")
39
except Exception as e:
40
logger.critical(f"Cannot discover bulbs: {str(e)}")
41
raise ValueError(f"Cannot discover bulbs: {str(e)}")
42
return __discovered
43
44
45
class MidiBulb(Y.Bulb):
46
47
@staticmethod48
def get_info_from_id(bulb_id: str) -> Tuple[str, str]:
49
"""
50
Returns IP and Firmware version of the bulb with given ID.51
"""52
for bulb in get_discovered():
53
if bulb[CAPABILITIES][ID] == bulb_id:
54
return bulb["ip"], bulb[CAPABILITIES]["fw_ver"]
55
logger.critical(
56
f"Bulb with ID {bulb_id} not found.\nRun wizard_configuration.py again.")
57
raise ValueError(
58
f"Bulb with ID {bulb_id} not found.\nRun wizard_configuration.py again.")
59
60
def __init__(self, bulb_id: str) -> None:
61
"""
62
MidiBuld class wraps around Yeelight's Bulb class, 63
providing additional functionality.\n64
It hides the IP address of the bulbs and allows65
to reach them just by their ID.66
67
:param bulb_id: ID of the bulb to connect to.68
"""69
bulb_ip, fw_ver = MidiBulb.get_info_from_id(bulb_id)
70
super().__init__(bulb_ip, effect="sudden")
71
self._ip: str = bulb_ip # TODO might not be needed
72
self._id: str = bulb_id
73
self._sticker_id: Optional[str] = None
74
self._channel: Optional[int] = None
75
self._fw_ver: Optional[str] = fw_ver
76
77
def __repr__(self) -> str:
78
s = f"MidiBulb: {self.id=}, {self._ip=}, {self.sticker_id=}, {self.channel=}\n"
79
return s
80
81
@property82
def id(self) -> str:
83
if self._id is None:
84
logger.error(f"Cannot get ID of bulb")
85
return "err"
86
return self._id
87
88
@property89
def sticker_id(self) -> str:
90
if self._sticker_id is None:
91
logger.error(f"Cannot get sticker ID of bulb")
92
return "err"
93
return self._sticker_id
94
95
@sticker_id.setter
96
def sticker_id(self, sticker_id: str) -> None:
97
self._sticker_id = sticker_id
98
return99
100
@property101
def channel(self) -> int:
102
if self._channel is None:
103
logger.error(f"Cannot get channel of bulb")
104
return -1
105
return self._channel
106
107
@channel.setter
108
def channel(self, ch: int) -> None:
109
if (ch < 0) or (ch > C.CHANNEL_COUNT - 1):
110
logger.error(
111
f"Invalid channel {ch} for bulb {self.id}. Expected 0..{C.CHANNEL_COUNT - 1}, saturating to {C.CHANNEL_COUNT - 1}.")
112
self._channel = C.CHANNEL_COUNT - 1
113
self._channel = ch
114
return115
116
@staticmethod117
def discover() -> List["MidiBulb"]:
118
"""
119
Discover available bulbs and return a list of MidiBulb objects.120
"""121
midi_bulbs = []
122
for yeelight_bulb in get_discovered():
123
midi_bulbs.append(MidiBulb(yeelight_bulb[CAPABILITIES][ID]))
124
return midi_bulbs
125
126
@contextmanager127
def distinguish(self) -> Generator[None, Any, None]:
128
"""
129
Helper function, that lights up the bulb for context.130
131
:note: Utilizes `with` statement.132
"""133
if self is None:
134
logger.error(f"Cannot distinguish bulb {self.id}")
135
return136
logger.info(
137
f"Distinguishing bulb {self.id} with sticker ID {self.sticker_id} in channel {self.channel}.")
138
self.turn_on()
139
self.set_rgb(*C.DISTINGUISH_COLOR)
140
self.set_brightness(100)
141
yield142
self.turn_off()
143
return144
145
146
def to_yaml(bulb_in_channel: Dict[int, List[MidiBulb]], filename: str = "config.yml") -> None:
147
yaml_formated = []
148
for bulbs in bulb_in_channel.values():
149
for bulb in bulbs:
150
yaml_formated.append(
151
{152
"bulb_id": bulb.id,
153
"sticker": bulb.sticker_id,
154
"channel": bulb.channel,
155
"firmware": bulb._fw_ver
156
}157
)158
with open(filename, "w+") as f:
159
yaml.dump(yaml_formated, f)
160
161
162
def from_yaml(filename: str = "config.yml") -> Dict[int, List[MidiBulb]]:
163
with open(filename, "r") as f:
164
yaml_formated = yaml.safe_load(f)
165
ret: Dict[int, List[MidiBulb]] = {}
166
for yaml_bulb in yaml_formated:
167
bulb_id = yaml_bulb["bulb_id"]
168
sticker_id = yaml_bulb["sticker"]
169
channel = int(yaml_bulb["channel"])
170
if channel not in ret.keys():
171
ret[channel] = []
172
midibulb = MidiBulb(bulb_id)
173
midibulb.sticker_id = sticker_id
174
midibulb.channel = channel
175
ret[channel].append(midibulb)
176
return ret
177
178
179
@contextmanager180
def distinguish_channel(bulbs: List[MidiBulb]) -> Generator[None, Any, None]:
181
for bulb in bulbs:
182
bulb.turn_on()
183
bulb.set_rgb(*C.DISTINGUISH_COLOR)
184
bulb.set_brightness(100)
185
yield186
for bulb in bulbs:
187
bulb.turn_off()