Merge d1c4e42 into b6e5b1a

README.md

@@ -53,6 +53,12 @@ apt-get install libusb-dev libudev-dev
 pip3 install blink1
 ```
 
+Install `yeelight` for Xiaomi Yeelight:
+
+```
+pip3 install yeelight
+```
+
 
 Usage
 -----
@@ -125,6 +131,20 @@ chroma-feedback --consumer=thingm_blink
 ```
 
 
+Xiaomi Yeelight
+---------------
+
+| Name     | Mandatory |
+|----------|-----------|
+| IP       | optional  |
+
+Indicate status via lights:
+
+```
+chroma-feedback --consumer=xiaomi_yeelight
+```
+
+
 Providers
 =========
 

chroma_feedback/consumer/__init__.py

@@ -4,5 +4,6 @@
 [
 	'philips_hue',
 	'razer_chroma',
-	'thingm_blink'
+	'thingm_blink',
+	'xiaomi_yeelight'
 ]

chroma_feedback/consumer/philips_hue/core.py

@@ -15,6 +15,8 @@ def init(program):
 		if ip:
 			program.add_argument('--philips-hue-ip', default = ip)
 		else:
+			print(wording.get('discovery_no').format('IP') + wording.get('exclamation_mark'))
+			print()
 			program.add_argument('--philips-hue-ip', required = True)
 		program.add_argument('--philips-hue-group', action = 'append')
 	args = program.parse_known_args()[0]

chroma_feedback/consumer/xiaomi_yeelight/__init__.py

@@ -0,0 +1 @@
+from .core import init, run, process

chroma_feedback/consumer/xiaomi_yeelight/core.py

@@ -0,0 +1,104 @@
+import socket
+from chroma_feedback import wording
+from .factory import bulb_factory
+
+args = None
+
+
+def init(program):
+	global args
+
+	if not args:
+		ip = discover_ips()
+
+		if ip:
+			program.add_argument('--xiaomi-yeelight-ip', default = ip)
+		else:
+			print(wording.get('discovery_no').format('IP') + wording.get('exclamation_mark'))
+			print()
+			program.add_argument('--xiaomi-yeelight-ip', required = True)
+	args = program.parse_known_args()[0]
+
+
+def run(status):
+	devices = []
+
+	for ip in args.xiaomi_yeelight_ip:
+		devices.append(bulb_factory(ip))
+	if not devices:
+		exit(wording.get('device_no') + wording.get('exclamation_mark'))
+	return process(status, devices)
+
+
+def process(status, devices):
+	result = []
+
+	# process devices
+
+	for device in devices:
+		device_name = device.get_properties()['name']
+
+		if status == 'passed':
+			result.append(
+			{
+				'consumer': 'xiaomi_yeelight',
+				'name': device_name,
+				'active': static(device, [0, 255, 0]),
+				'status': status
+			})
+		if status == 'process':
+			result.append(
+			{
+				'consumer': 'xiaomi_yeelight',
+				'name': device_name,
+				'active': static(device, [255, 255, 0]),
+				'status': status
+			})
+		if status == 'errored':
+			result.append(
+			{
+				'consumer': 'xiaomi_yeelight',
+				'name': device_name,
+				'active': pulsate(device, [255, 255, 255]),
+				'status': status
+			})
+		if status == 'failed':
+			result.append(
+			{
+				'consumer': 'xiaomi_yeelight',
+				'name': device_name,
+				'active': pulsate(device, [255, 0, 0]),
+				'status': status
+			})
+	return result
+
+
+def static(device, rgb):
+	return device.set_rgb(rgb[0], rgb[1], rgb[2]) == 'ok'
+
+
+def pulsate(device, rgb):
+	return device.set_rgb(rgb[0], rgb[1], rgb[2]) == 'ok'
+
+
+def discover_ips():
+	message =\
+	[
+		'M-SEARCH * HTTP/1.1',
+		'HOST: 239.255.255.250:1982',
+		'MAN: "ssdp:discover"',
+		'ST: wifi_bulb'
+	]
+	SOCKET = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+	SOCKET.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 32)
+	SOCKET.settimeout(1)
+	SOCKET.sendto('\r\n'.join(message).encode(), ('239.255.255.250', 1982))
+	ips = []
+
+	while True:
+		try:
+			ip = SOCKET.recvfrom(65507)[1][0]
+		except socket.timeout:
+			break
+		ips.append(ip)
+	return list(dict.fromkeys(ips))

chroma_feedback/consumer/xiaomi_yeelight/factory.py

@@ -0,0 +1,21 @@
+from chroma_feedback import wording
+
+
+def bulb_factory(ip):
+	bulb = None
+
+	# handle import
+
+	try:
+		from yeelight import Bulb, BulbException
+	except ImportError:
+		exit(wording.get('package_no').format('YEELIGHT') + wording.get('exclamation_mark'))
+
+	# create instance
+
+	try:
+		bulb = Bulb(ip)
+		bulb.turn_on()
+	except BulbException:
+		exit(wording.get('connection_no').format('YEELIGHT') + wording.get('exclamation_mark'))
+	return bulb

chroma_feedback/wording.py

@@ -3,6 +3,7 @@
 	'version_no': 'Python {}.{} not supported',
 	'package_no': 'Package {} not found',
 	'connection_no': 'Connection to {} not found',
+	'discovery_no': 'Discovery for {} failed',
 	'daemon_no': 'Daemon for {} not found',
 	'device_no': 'Device not found',
 	'group_no': 'Group not found',

setup.py

@@ -33,7 +33,8 @@
 		'chroma_feedback.consumer',
 		'chroma_feedback.consumer.philips_hue',
 		'chroma_feedback.consumer.razer_chroma',
-		'chroma_feedback.consumer.thingm_blink'
+		'chroma_feedback.consumer.thingm_blink',
+		'chroma_feedback.consumer.xiaomi_yeelight'
 	],
 	scripts =
 	[

tests/consumer/xiaomi_yeelight/test_core.py

@@ -0,0 +1,40 @@
+from mock import MagicMock
+from chroma_feedback.consumer import xiaomi_yeelight
+
+mock = MagicMock()
+devices =\
+[
+	mock
+]
+
+
+def test_process_passed():
+	result = xiaomi_yeelight.process('passed', devices)
+
+	assert result[0]['consumer'] == 'xiaomi_yeelight'
+	assert result[0]['name']
+	assert result[0]['status'] == 'passed'
+
+
+def test_process_process():
+	result = xiaomi_yeelight.process('process', devices)
+
+	assert result[0]['consumer'] == 'xiaomi_yeelight'
+	assert result[0]['name']
+	assert result[0]['status'] == 'process'
+
+
+def test_process_errored():
+	result = xiaomi_yeelight.process('errored', devices)
+
+	assert result[0]['consumer'] == 'xiaomi_yeelight'
+	assert result[0]['name']
+	assert result[0]['status'] == 'errored'
+
+
+def test_process_failed():
+	result = xiaomi_yeelight.process('failed', devices)
+
+	assert result[0]['consumer'] == 'xiaomi_yeelight'
+	assert result[0]['name']
+	assert result[0]['status'] == 'failed'