Skip to content

Commit

Permalink
Merge 0812249 into 46ca51f
Browse files Browse the repository at this point in the history
  • Loading branch information
henryruhs committed Oct 1, 2022
2 parents 46ca51f + 0812249 commit 64d265f
Show file tree
Hide file tree
Showing 16 changed files with 248 additions and 11 deletions.
2 changes: 1 addition & 1 deletion .github/terminal-session.svg
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
1 change: 1 addition & 0 deletions chroma_feedback/consumer/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
__all__ =\
[
'agile_innovative_blinkstick',
'compulab_fit_statusb',
'elgato_streamdeck',
'embrava_blynclight',
'kuando_busylight',
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@


def support() -> bool:
return helper.is_linux() is True or helper.is_mac() is True
return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True


def init(program : ArgumentParser) -> None:
Expand Down
1 change: 1 addition & 0 deletions chroma_feedback/consumer/compulab_fit_statusb/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
from .core import init, run, support
30 changes: 30 additions & 0 deletions chroma_feedback/consumer/compulab_fit_statusb/api.py
Original file line number Diff line number Diff line change
@@ -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 USBLightIOError, USBLightNotFound
from busylight.lights.compulap import Fit_StatUSB as api

try:
api.first_light().release()
except (USBLightIOError, USBLightNotFound):
logger.error(wording.get('connection_not_found').format('compulab_fit_statusb') + 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()
30 changes: 30 additions & 0 deletions chroma_feedback/consumer/compulab_fit_statusb/core.py
Original file line number Diff line number Diff line change
@@ -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('--compulab-fit-statusb-light-id', action = 'append')
ARGS = helper.get_first(program.parse_known_args())


def run(producer_report : List[ProducerReport]) -> List[Consumer]:
light = filter_lights(get_lights(), ARGS.compulab_fit_statusb_light_id)

if not light:
logger.error(wording.get('device_not_found') + wording.get('exclamation_mark'))
sys.exit()
return process_lights(light, producer_report)
48 changes: 48 additions & 0 deletions chroma_feedback/consumer/compulab_fit_statusb/light.py
Original file line number Diff line number Diff line change
@@ -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': 'compulab_fit_statusb',
'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

2 changes: 1 addition & 1 deletion chroma_feedback/consumer/embrava_blynclight/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@


def support() -> bool:
return helper.is_linux() is True or helper.is_mac() is True
return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True


def init(program : ArgumentParser) -> None:
Expand Down
2 changes: 1 addition & 1 deletion chroma_feedback/consumer/kuando_busylight/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@


def support() -> bool:
return helper.is_linux() is True or helper.is_mac() is True
return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True


def init(program : ArgumentParser) -> None:
Expand Down
2 changes: 1 addition & 1 deletion chroma_feedback/consumer/luxafor_flag/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ def api_factory() -> Any:
try:
api.first_light().release()
except (USBLightIOError, USBLightNotFound):
logger.error(wording.get('connection_not_found').format('lifx_light') + wording.get('exclamation_mark'))
logger.error(wording.get('connection_not_found').format('luxafor_flag') + wording.get('exclamation_mark'))
sys.exit()
return api
except ImportError:
Expand Down
2 changes: 1 addition & 1 deletion chroma_feedback/consumer/luxafor_flag/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@


def support() -> bool:
return helper.is_linux() is True or helper.is_mac() is True
return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True


def init(program : ArgumentParser) -> None:
Expand Down
2 changes: 1 addition & 1 deletion chroma_feedback/consumer/thingm_blink1/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@


def support() -> bool:
return helper.is_linux() is True or helper.is_mac() is True
return helper.is_linux() is True or helper.is_mac() is True or helper.is_windows() is True


def init(program : ArgumentParser) -> None:
Expand Down
2 changes: 1 addition & 1 deletion chroma_feedback/metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
{
'name': 'chroma-feedback',
'description': 'Turn your RGB powered hardware into a status indicator for continuous integration, continuous deployment and infrastructure monitoring',
'version': '12.3.0',
'version': '13.0.0-next.1',
'license': 'MIT',
'keywords': ' '.join(producer.__all__),
'author': 'Henry Ruhs',
Expand Down
7 changes: 4 additions & 3 deletions setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
'chroma_feedback.producer.vercel',
'chroma_feedback.consumer',
'chroma_feedback.consumer.agile_innovative_blinkstick',
'chroma_feedback.consumer.compulab_fit_statusb',
'chroma_feedback.consumer.elgato_streamdeck',
'chroma_feedback.consumer.embrava_blynclight',
'chroma_feedback.consumer.kuando_busylight',
Expand All @@ -64,15 +65,15 @@
},
install_requires =
[
'busylight-for-humans==0.21.1',
'busylight-for-humans==0.22.2',
'lifxlan==1.2.7',
'nanoleafapi==2.1.1',
'nanoleafapi==2.1.2',
'phue==1.1',
'python-magichue==0.3.2',
'pyqt5==5.15.7',
'pywizlight==0.5.14',
'requests==2.28.1',
'streamdeck==0.9.1',
'streamdeck==0.9.2',
'yeelight==0.7.10'
],
tests_require =
Expand Down
Empty file.
126 changes: 126 additions & 0 deletions tests/consumer/compulab_fit_statusb/test_device.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
import pytest
from unittest.mock import MagicMock

from chroma_feedback.consumer.compulab_fit_statusb.light import process_lights

MOCK = MagicMock()


def test_process_passed() -> None:
try:
result = process_lights(
{
MOCK
},
[
{
'name': 'github',
'symbol': None,
'message': None,
'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
'status': 'passed'
}
])

assert result[0]['name'] == 'compulab_fit_statusb'
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_lights(
{
MOCK
},
[
{
'name': 'github',
'symbol': None,
'message': None,
'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
'status': 'started'
}
])

assert result[0]['name'] == 'compulab_fit_statusb'
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_lights(
{
MOCK
},
[
{
'name': 'github',
'symbol': None,
'message': None,
'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
'status': 'errored'
}
])

assert result[0]['name'] == 'compulab_fit_statusb'
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_lights(
{
MOCK
},
[
{
'name': 'github',
'symbol': None,
'message': None,
'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
'status': 'warned'
}
])

assert result[0]['name'] == 'compulab_fit_statusb'
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_lights(
{
MOCK
},
[
{
'name': 'github',
'symbol': None,
'message': None,
'url': 'https://github.com/henryruhs/chroma-feedback/actions/runs/1',
'status': 'failed'
}
])

assert result[0]['name'] == 'compulab_fit_statusb'
assert result[0]['type'] == 'device'
assert result[0]['description']
assert result[0]['status'] == 'failed'
except:
pytest.skip()

0 comments on commit 64d265f

Please sign in to comment.