Support for Gigabyte RGB Fusion 2.0

chroma_feedback/consumer/__init__.py

@@ -8,6 +8,7 @@
 	'embrava.blynclight': 'chroma_feedback.consumer.embrava.blynclight',
 	'embrava.blynclight_mini': 'chroma_feedback.consumer.embrava.blynclight_mini',
 	'embrava.blynclight_plus': 'chroma_feedback.consumer.embrava.blynclight_plus',
+	'gigabyte.rgb_fusion2': 'chroma_feedback.consumer.gigabyte.rgb_fusion2',
 	'kuando.busylight_alpha': 'chroma_feedback.consumer.kuando.busylight_alpha',
 	'kuando.busylight_omega': 'chroma_feedback.consumer.kuando.busylight_omega',
 	'lifx': 'chroma_feedback.consumer.lifx.lifx',

chroma_feedback/consumer/gigabyte/rgb_fusion2/__init__.py

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

chroma_feedback/consumer/gigabyte/rgb_fusion2/api.py

@@ -0,0 +1,24 @@
+import sys
+from typing import Any
+
+from chroma_feedback import logger, helper, wording
+
+API = None
+
+
+def get_api() -> Any:
+	global API
+
+	if not API:
+		API = api_factory()
+	return API
+
+
+def api_factory() -> Any:
+	try:
+		from liquidctl.driver.rgb_fusion2 import RgbFusion2 as api
+
+		return api
+	except ImportError:
+		logger.error(wording.get('package_not_found').format('liquidctl') + wording.get('exclamation_mark'))
+		sys.exit()

chroma_feedback/consumer/gigabyte/rgb_fusion2/core.py

@@ -0,0 +1,28 @@
+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 .device import get_devices, process_devices
+
+ARGS = None
+
+
+def support() -> bool:
+	return helper.is_linux() is True or helper.is_windows() is True
+
+
+def init(program : ArgumentParser) -> None:
+	global ARGS
+
+	ARGS = helper.get_first(program.parse_known_args())
+
+
+def run(producer_report : List[ProducerReport]) -> List[Consumer]:
+	devices = get_devices()
+
+	if not devices:
+		logger.error(wording.get('device_not_found') + wording.get('exclamation_mark'))
+		sys.exit()
+	return process_devices(devices, producer_report)

chroma_feedback/consumer/gigabyte/rgb_fusion2/device.py

@@ -0,0 +1,45 @@
+import atexit
+from typing import Any, List
+
+from chroma_feedback import color, reporter
+from chroma_feedback.typing import Color, Consumer, ProducerReport, Status
+from .api import get_api
+
+DEVICES = None
+
+
+def get_devices() -> Any:
+	global DEVICES
+
+	if not DEVICES:
+		DEVICES = get_api().find_supported_devices()
+	return DEVICES
+
+
+def process_devices(devices : Any, producer_report : List[ProducerReport]) -> List[Consumer]:
+	result : List[Consumer] = []
+	status : Status = reporter.resolve_report_status(producer_report)
+
+	for device in devices:
+		with device.connect():
+			set_device(device, color.get_by_status(status))
+			register_reset_device(device)
+			result.append(
+			{
+				'name': 'gigabyte.rgb_fusion2',
+				'type': 'device',
+				'description': device.description,
+				'status': status
+			})
+	return result
+
+
+def set_device(device : Any, color_config : Color) -> None:
+	device.set_color('sync', 'fixed',
+	[
+		tuple(color_config['rgb'])
+	])
+
+
+def register_reset_device(device : Any) -> None:
+	atexit.register(lambda: set_device(device, color.get_reset()))

setup.py

@@ -43,6 +43,7 @@
 		'chroma_feedback.consumer.embrava.blynclight',
 		'chroma_feedback.consumer.embrava.blynclight_mini',
 		'chroma_feedback.consumer.embrava.blynclight_plus',
+		'chroma_feedback.consumer.gigabyte.rgb_fusion2',
 		'chroma_feedback.consumer.kuando.busylight_alpha',
 		'chroma_feedback.consumer.kuando.busylight_omega',
 		'chroma_feedback.consumer.lifx.lifx',
@@ -76,6 +77,7 @@
 	[
 		'busylight-for-humans==0.25.5',
 		'lifxlan==1.2.7',
+		'liquidctl==1.12.1',
 		'nanoleafapi==2.1.2',
 		'phue==1.1',
 		'python-magichue==0.3.2',

tests/consumer/gigabyte/rgb_fusion2/test_device.py

@@ -0,0 +1,126 @@
+import pytest
+from unittest.mock import MagicMock
+
+from chroma_feedback.consumer.gigabyte.rgb_fusion2.device import process_devices
+
+MOCK = MagicMock()
+
+
+def test_process_passed() -> None:
+	try:
+		result = process_devices(
+		{
+			MOCK
+		},
+		[
+			{
+				'name': 'microsoft.github',
+				'symbol': None,
+				'message': None,
+				'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
+				'status': 'passed'
+			}
+		])
+
+		assert result[0]['name'] == 'gigabyte.rgb_fusion2'
+		assert result[0]['type'] == 'device'
+		assert result[0]['description']
+		assert result[0]['status'] == 'passed'
+	except:
+		pytest.skip()
+
+
+def test_process_started() -> None:
+	try:
+		result = process_devices(
+		{
+			MOCK
+		},
+		[
+			{
+				'name': 'microsoft.github',
+				'symbol': None,
+				'message': None,
+				'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
+				'status': 'started'
+			}
+		])
+
+		assert result[0]['name'] == 'gigabyte.rgb_fusion2'
+		assert result[0]['type'] == 'device'
+		assert result[0]['description']
+		assert result[0]['status'] == 'started'
+	except:
+		pytest.skip()
+
+
+def test_process_errored() -> None:
+	try:
+		result = process_devices(
+		{
+			MOCK
+		},
+		[
+			{
+				'name': 'microsoft.github',
+				'symbol': None,
+				'message': None,
+				'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
+				'status': 'errored'
+			}
+		])
+
+		assert result[0]['name'] == 'gigabyte.rgb_fusion2'
+		assert result[0]['type'] == 'device'
+		assert result[0]['description']
+		assert result[0]['status'] == 'errored'
+	except:
+		pytest.skip()
+
+
+def test_process_warned() -> None:
+	try:
+		result = process_devices(
+		{
+			MOCK
+		},
+		[
+			{
+				'name': 'microsoft.github',
+				'symbol': None,
+				'message': None,
+				'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
+				'status': 'warned'
+			}
+		])
+
+		assert result[0]['name'] == 'gigabyte.rgb_fusion2'
+		assert result[0]['type'] == 'device'
+		assert result[0]['description']
+		assert result[0]['status'] == 'warned'
+	except:
+		pytest.skip()
+
+
+def test_process_failed() -> None:
+	try:
+		result = process_devices(
+		{
+			MOCK
+		},
+		[
+			{
+				'name': 'microsoft.github',
+				'symbol': None,
+				'message': None,
+				'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
+				'status': 'failed'
+			}
+		])
+
+		assert result[0]['name'] == 'gigabyte.rgb_fusion2'
+		assert result[0]['type'] == 'device'
+		assert result[0]['description']
+		assert result[0]['status'] == 'failed'
+	except:
+		pytest.skip()