Implement all devices part3

chroma_feedback/consumer/__init__.py

@@ -15,6 +15,9 @@
 	'luxafor.mute': 'chroma_feedback.consumer.luxafor.mute',
 	'luxafor.orb': 'chroma_feedback.consumer.luxafor.orb',
 	'magic.hue': 'chroma_feedback.consumer.magic.hue',
+	'muteme': 'chroma_feedback.consumer.muteme.muteme',
+	'muteme.muteme_mini': 'chroma_feedback.consumer.muteme.muteme_mini',
+	'mutesync': 'chroma_feedback.consumer.mutesync.mutesync',
 	'nanoleaf': 'chroma_feedback.consumer.nanoleaf.nanoleaf',
 	'philips.hue': 'chroma_feedback.consumer.philips.hue',
 	'plantronics.status_indicator': 'chroma_feedback.consumer.plantronics.status_indicator',

chroma_feedback/consumer/muteme/muteme/__init__.py

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

chroma_feedback/consumer/muteme/muteme/api.py

@@ -0,0 +1,30 @@
+import sys
+from typing import Any
+
+from chroma_feedback import logger, wording
+
+API = None
+
+
+def get_api() -> Any:
+	global API
+
+	if not API:
+		API = api_factory()
+	return API
+
+
+def api_factory() -> Any:
+	try:
+		from busylight.lights import NoLightsFound, LightUnavailable
+		from busylight.lights.muteme import MuteMe as api
+
+		try:
+			api.first_light().acquire()
+		except (NoLightsFound, LightUnavailable):
+			logger.error(wording.get('connection_not_found').format('muteme') + wording.get('exclamation_mark'))
+			sys.exit()
+		return api
+	except ImportError:
+		logger.error(wording.get('package_not_found').format('busylight-for-humans') + wording.get('exclamation_mark'))
+		sys.exit()

chroma_feedback/consumer/muteme/muteme/core.py

@@ -0,0 +1,30 @@
+import sys
+from argparse import ArgumentParser
+from typing import List
+
+from chroma_feedback import helper, logger, wording
+from chroma_feedback.typing import Consumer, ProducerReport
+from .light import filter_lights, get_lights, process_lights
+
+ARGS = None
+
+
+def support() -> bool:
+	return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True
+
+
+def init(program : ArgumentParser) -> None:
+	global ARGS
+
+	if not ARGS:
+		program.add_argument('--muteme-light-id', action = 'append')
+	ARGS = helper.get_first(program.parse_known_args())
+
+
+def run(producer_report : List[ProducerReport]) -> List[Consumer]:
+	lights = filter_lights(get_lights(), ARGS.muteme_light_id)
+
+	if not lights:
+		logger.error(wording.get('light_not_found') + wording.get('exclamation_mark'))
+		sys.exit()
+	return process_lights(lights, producer_report)

chroma_feedback/consumer/muteme/muteme/light.py

@@ -0,0 +1,48 @@
+import copy
+from typing import Any, List
+
+from chroma_feedback import color, helper, reporter
+from chroma_feedback.typing import Color, Consumer, ProducerReport, Status
+from .api import get_api
+
+LIGHTS = None
+
+
+def get_lights() -> Any:
+	global LIGHTS
+
+	if not LIGHTS:
+		LIGHTS = get_api().all_lights()
+	return LIGHTS
+
+
+def filter_lights(lights : Any, light_ids : List[str]) -> Any:
+	if light_ids:
+		for light in copy.copy(lights):
+			if light.path not in light_ids:
+				lights.remove(light)
+	return lights
+
+
+def process_lights(lights : Any, producer_report : List[ProducerReport]) -> List[Consumer]:
+	result : List[Consumer] = []
+	status : Status = reporter.resolve_report_status(producer_report)
+
+	# process lights
+
+	for light in lights:
+		if set_light(light, color.get_by_status(status)):
+			result.append(
+			{
+				'name': 'mutemer',
+				'type': 'light',
+				'description': helper.create_description(light.info['product_string'], light.path),
+				'status': status
+			})
+	return result
+
+
+def set_light(light : Any, color_config : Color) -> bool:
+	light.on(tuple(color_config['rgb']))
+	return light.is_on is True
+

chroma_feedback/consumer/muteme/muteme_mini/__init__.py

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

chroma_feedback/consumer/muteme/muteme_mini/api.py

@@ -0,0 +1,30 @@
+import sys
+from typing import Any
+
+from chroma_feedback import logger, wording
+
+API = None
+
+
+def get_api() -> Any:
+	global API
+
+	if not API:
+		API = api_factory()
+	return API
+
+
+def api_factory() -> Any:
+	try:
+		from busylight.lights import NoLightsFound, LightUnavailable
+		from busylight.lights.muteme import MuteMe_Mini as api
+
+		try:
+			api.first_light().acquire()
+		except (NoLightsFound, LightUnavailable):
+			logger.error(wording.get('connection_not_found').format('muteme.muteme_mini') + wording.get('exclamation_mark'))
+			sys.exit()
+		return api
+	except ImportError:
+		logger.error(wording.get('package_not_found').format('busylight-for-humans') + wording.get('exclamation_mark'))
+		sys.exit()

chroma_feedback/consumer/muteme/muteme_mini/core.py

@@ -0,0 +1,30 @@
+import sys
+from argparse import ArgumentParser
+from typing import List
+
+from chroma_feedback import helper, logger, wording
+from chroma_feedback.typing import Consumer, ProducerReport
+from .light import filter_lights, get_lights, process_lights
+
+ARGS = None
+
+
+def support() -> bool:
+	return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True
+
+
+def init(program : ArgumentParser) -> None:
+	global ARGS
+
+	if not ARGS:
+		program.add_argument('--muteme-muteme-mini-light-id', action = 'append')
+	ARGS = helper.get_first(program.parse_known_args())
+
+
+def run(producer_report : List[ProducerReport]) -> List[Consumer]:
+	lights = filter_lights(get_lights(), ARGS.muteme_muteme_mini_light_id)
+
+	if not lights:
+		logger.error(wording.get('light_not_found') + wording.get('exclamation_mark'))
+		sys.exit()
+	return process_lights(lights, producer_report)

chroma_feedback/consumer/muteme/muteme_mini/light.py

@@ -0,0 +1,48 @@
+import copy
+from typing import Any, List
+
+from chroma_feedback import color, helper, reporter
+from chroma_feedback.typing import Color, Consumer, ProducerReport, Status
+from .api import get_api
+
+LIGHTS = None
+
+
+def get_lights() -> Any:
+	global LIGHTS
+
+	if not LIGHTS:
+		LIGHTS = get_api().all_lights()
+	return LIGHTS
+
+
+def filter_lights(lights : Any, light_ids : List[str]) -> Any:
+	if light_ids:
+		for light in copy.copy(lights):
+			if light.path not in light_ids:
+				lights.remove(light)
+	return lights
+
+
+def process_lights(lights : Any, producer_report : List[ProducerReport]) -> List[Consumer]:
+	result : List[Consumer] = []
+	status : Status = reporter.resolve_report_status(producer_report)
+
+	# process lights
+
+	for light in lights:
+		if set_light(light, color.get_by_status(status)):
+			result.append(
+			{
+				'name': 'muteme.muteme_mini',
+				'type': 'light',
+				'description': helper.create_description(light.info['product_string'], light.path),
+				'status': status
+			})
+	return result
+
+
+def set_light(light : Any, color_config : Color) -> bool:
+	light.on(tuple(color_config['rgb']))
+	return light.is_on is True
+

chroma_feedback/consumer/mutesync/mutesync/__init__.py

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

chroma_feedback/consumer/mutesync/mutesync/api.py

@@ -0,0 +1,30 @@
+import sys
+from typing import Any
+
+from chroma_feedback import logger, wording
+
+API = None
+
+
+def get_api() -> Any:
+	global API
+
+	if not API:
+		API = api_factory()
+	return API
+
+
+def api_factory() -> Any:
+	try:
+		from busylight.lights import NoLightsFound, LightUnavailable
+		from busylight.lights.mutesync import MuteSync as api
+
+		try:
+			api.first_light().acquire()
+		except (NoLightsFound, LightUnavailable):
+			logger.error(wording.get('connection_not_found').format('mutesync') + wording.get('exclamation_mark'))
+			sys.exit()
+		return api
+	except ImportError:
+		logger.error(wording.get('package_not_found').format('busylight-for-humans') + wording.get('exclamation_mark'))
+		sys.exit()

chroma_feedback/consumer/mutesync/mutesync/core.py

@@ -0,0 +1,30 @@
+import sys
+from argparse import ArgumentParser
+from typing import List
+
+from chroma_feedback import helper, logger, wording
+from chroma_feedback.typing import Consumer, ProducerReport
+from .light import filter_lights, get_lights, process_lights
+
+ARGS = None
+
+
+def support() -> bool:
+	return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True
+
+
+def init(program : ArgumentParser) -> None:
+	global ARGS
+
+	if not ARGS:
+		program.add_argument('--mutesync-light-id', action = 'append')
+	ARGS = helper.get_first(program.parse_known_args())
+
+
+def run(producer_report : List[ProducerReport]) -> List[Consumer]:
+	lights = filter_lights(get_lights(), ARGS.mutesync_light_id)
+
+	if not lights:
+		logger.error(wording.get('light_not_found') + wording.get('exclamation_mark'))
+		sys.exit()
+	return process_lights(lights, producer_report)

chroma_feedback/consumer/mutesync/mutesync/light.py

@@ -0,0 +1,48 @@
+import copy
+from typing import Any, List
+
+from chroma_feedback import color, helper, reporter
+from chroma_feedback.typing import Color, Consumer, ProducerReport, Status
+from .api import get_api
+
+LIGHTS = None
+
+
+def get_lights() -> Any:
+	global LIGHTS
+
+	if not LIGHTS:
+		LIGHTS = get_api().all_lights()
+	return LIGHTS
+
+
+def filter_lights(lights : Any, light_ids : List[str]) -> Any:
+	if light_ids:
+		for light in copy.copy(lights):
+			if light.path not in light_ids:
+				lights.remove(light)
+	return lights
+
+
+def process_lights(lights : Any, producer_report : List[ProducerReport]) -> List[Consumer]:
+	result : List[Consumer] = []
+	status : Status = reporter.resolve_report_status(producer_report)
+
+	# process lights
+
+	for light in lights:
+		if set_light(light, color.get_by_status(status)):
+			result.append(
+			{
+				'name': 'mutesync',
+				'type': 'light',
+				'description': helper.create_description(light.info['product_string'], light.path),
+				'status': status
+			})
+	return result
+
+
+def set_light(light : Any, color_config : Color) -> bool:
+	light.on(tuple(color_config['rgb']))
+	return light.is_on is True
+

chroma_feedback/typing.py

@@ -44,6 +44,9 @@
 	'luxafor.mute',
 	'luxafor.orb',
 	'magic.hue',
+	'muteme',
+	'muteme.muteme_mini',
+	'mutesync',
 	'nanoleaf',
 	'philips.hue',
 	'plantronics.status_indicator',

setup.py

@@ -50,6 +50,9 @@
 		'chroma_feedback.consumer.luxafor.mute',
 		'chroma_feedback.consumer.luxafor.orb',
 		'chroma_feedback.consumer.magic.hue',
+		'chroma_feedback.consumer.muteme.muteme',
+		'chroma_feedback.consumer.muteme.muteme_mini',
+		'chroma_feedback.consumer.mutesync.mutesync',
 		'chroma_feedback.consumer.nanoleaf.nanoleaf',
 		'chroma_feedback.consumer.philips.hue',
 		'chroma_feedback.consumer.plantronics.status_indicator',