diff --git a/chroma_feedback/consumer/__init__.py b/chroma_feedback/consumer/__init__.py index 0f20fa0..580e098 100644 --- a/chroma_feedback/consumer/__init__.py +++ b/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', diff --git a/chroma_feedback/consumer/gigabyte/__init__.py b/chroma_feedback/consumer/gigabyte/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/chroma_feedback/consumer/gigabyte/rgb_fusion2/__init__.py b/chroma_feedback/consumer/gigabyte/rgb_fusion2/__init__.py new file mode 100644 index 0000000..34dcffe --- /dev/null +++ b/chroma_feedback/consumer/gigabyte/rgb_fusion2/__init__.py @@ -0,0 +1 @@ +from .core import init, run, support diff --git a/chroma_feedback/consumer/gigabyte/rgb_fusion2/api.py b/chroma_feedback/consumer/gigabyte/rgb_fusion2/api.py new file mode 100644 index 0000000..cf25149 --- /dev/null +++ b/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() diff --git a/chroma_feedback/consumer/gigabyte/rgb_fusion2/core.py b/chroma_feedback/consumer/gigabyte/rgb_fusion2/core.py new file mode 100644 index 0000000..4ac40c0 --- /dev/null +++ b/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) diff --git a/chroma_feedback/consumer/gigabyte/rgb_fusion2/device.py b/chroma_feedback/consumer/gigabyte/rgb_fusion2/device.py new file mode 100644 index 0000000..31647ea --- /dev/null +++ b/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())) diff --git a/setup.py b/setup.py index 0a1d427..a0caeb9 100755 --- a/setup.py +++ b/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', diff --git a/tests/consumer/gigabyte/__init__.py b/tests/consumer/gigabyte/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/consumer/gigabyte/rgb_fusion2/__init__.py b/tests/consumer/gigabyte/rgb_fusion2/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/consumer/gigabyte/rgb_fusion2/test_device.py b/tests/consumer/gigabyte/rgb_fusion2/test_device.py new file mode 100644 index 0000000..4947f51 --- /dev/null +++ b/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()