generalized main - get values from config
This commit is contained in:
parent
823552ce7f
commit
da9373e74d
|
@ -1,4 +1,5 @@
|
||||||
generated_3d
|
generated_3d
|
||||||
|
venv
|
||||||
build
|
build
|
||||||
*.FCStd1
|
*.FCStd1
|
||||||
*.blend1
|
*.blend1
|
||||||
|
|
|
@ -39,6 +39,13 @@ mouse_led_effect_to_message_id = {
|
||||||
EffectReverseSwipe: 10,
|
EffectReverseSwipe: 10,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
mouse_leds_index_ranges = {
|
||||||
|
TouchButton.RIGHT_FOOT: (0, 6),
|
||||||
|
TouchButton.LEFT_FOOT: (6, 6 + 6),
|
||||||
|
TouchButton.LEFT_EAR: (6 + 6, 6 + 6 + 16),
|
||||||
|
TouchButton.RIGHT_EAR: (6 + 6 + 16, 6 + 6 + 16 + 17),
|
||||||
|
}
|
||||||
|
|
||||||
PREV_BUTTON_LED_MSG = 20
|
PREV_BUTTON_LED_MSG = 20
|
||||||
NEXT_BUTTON_LED_MSG = 21
|
NEXT_BUTTON_LED_MSG = 21
|
||||||
|
|
||||||
|
|
|
@ -1,4 +1,7 @@
|
||||||
|
#!/usr/bin/env python
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import sys
|
||||||
import serial_asyncio
|
import serial_asyncio
|
||||||
from led_cmds import (ColorRGBW, ColorHSV, EffectStaticConfig,
|
from led_cmds import (ColorRGBW, ColorHSV, EffectStaticConfig,
|
||||||
EffectRandomTwoColorInterpolationConfig, EffectAlexaSwipeConfig,
|
EffectRandomTwoColorInterpolationConfig, EffectAlexaSwipeConfig,
|
||||||
|
@ -9,175 +12,201 @@ from glob import glob
|
||||||
from copy import deepcopy
|
from copy import deepcopy
|
||||||
import os
|
import os
|
||||||
from hass_client import HomeAssistantClient
|
from hass_client import HomeAssistantClient
|
||||||
|
import argparse
|
||||||
|
from ruamel.yaml import YAML
|
||||||
|
import warnings
|
||||||
|
from pprint import pprint
|
||||||
|
|
||||||
HASS_URL = "https://ha.bauer.tech"
|
yaml = YAML(typ='safe')
|
||||||
HASS_TOKEN = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiI3YmY1OTc3NWJhZGI0MzFkOTBiNDBkZDA3OGQ1MTMxNSIsImlhdCI6MTY0MDAzNzI3MSwiZXhwIjoxOTU1Mzk3MjcxfQ.c7RPRP_hxzIwd3xcFTNLz94rOjLQDR0elH8RE-jCDgc"
|
|
||||||
MUSIC_FOLDER = "/home/martin/code/musicmouse/espmusicmouse/host_driver/music"
|
|
||||||
|
|
||||||
audio_player = AudioPlayer()
|
OFF_COLOR = ColorRGBW(0, 0, 0, 0)
|
||||||
current_figure = None
|
|
||||||
last_figure = None
|
|
||||||
|
|
||||||
rfid_token_map = {
|
|
||||||
bytes.fromhex("88041174e9"): "elefant",
|
|
||||||
bytes.fromhex("8804ce7230"): "fuchs",
|
|
||||||
bytes.fromhex("88040d71f0"): "eule",
|
|
||||||
bytes.fromhex("88043c6ede"): "omnom",
|
|
||||||
bytes.fromhex("88040b78ff"): "eichhoernchen",
|
|
||||||
bytes.fromhex("8804bc7444"): "hund",
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def parse_hex_color(color_str: str):
|
def parse_color(color_str: str):
|
||||||
if isinstance(color_str, ColorRGBW):
|
if isinstance(color_str, ColorRGBW):
|
||||||
return color_str
|
return color_str
|
||||||
color_str = color_str.lstrip('#')
|
elif color_str.startswith("#"):
|
||||||
t = tuple(int(color_str[i:i + 2], 16) / 255 for i in (0, 2, 4))
|
color_str = color_str.lstrip('#')
|
||||||
return ColorRGBW(*t, 0)
|
t = tuple(int(color_str[i:i + 2], 16) / 255 for i in (0, 2, 4))
|
||||||
|
return ColorRGBW(*t, 0)
|
||||||
|
elif color_str.startswith("w"):
|
||||||
|
color_str = color_str.lstrip("w")
|
||||||
|
return ColorRGBW(0, 0, 0, int(color_str, 16) / 255)
|
||||||
|
|
||||||
|
|
||||||
class FigureColorCfg:
|
def load_config(config_path):
|
||||||
def __init__(self, primary, secondary, background, accent):
|
with open(os.path.join(config_path, "config.yml")) as cfg_file:
|
||||||
self.primary = parse_hex_color(primary)
|
cfg = yaml.load(cfg_file)
|
||||||
self.secondary = parse_hex_color(secondary)
|
for figure_name, figure_cfg in cfg["figures"].items():
|
||||||
self.background = parse_hex_color(background)
|
figure_cfg["colors"] = [parse_color(c) for c in figure_cfg["colors"]]
|
||||||
self.accent = parse_hex_color(accent)
|
if 'media_files' not in figure_cfg:
|
||||||
|
figure_cfg['media_files'] = sorted(glob(os.path.join(config_path, figure_name)))
|
||||||
|
return cfg
|
||||||
|
|
||||||
|
|
||||||
color_cfg = {
|
class MusicMouseState:
|
||||||
"elefant": FigureColorCfg("#ffff00", "#00c8ff", "#094b46", "#c20099"),
|
def __init__(self, protocol: MusicMouseProtocol):
|
||||||
"fuchs": FigureColorCfg("#F4D35E", "#F95738", "#F95738", "#083d77"),
|
self.current_figure: str = None
|
||||||
"omnom": FigureColorCfg("#005102", "#3bc405", "#005102", "#3bc405"),
|
self.last_figure: str = None
|
||||||
"eichhoernchen": FigureColorCfg("#ff0ada", "#4BC6B9", "#69045a", "#4BC6B9"),
|
self.current_mouse_led_effect = None
|
||||||
"hund": FigureColorCfg("#ffff00", "#00c8ff", "#094b46", "#c20099"),
|
self.current_led_ring_effect = None
|
||||||
"eule": FigureColorCfg("#e5a200", "#f8e300", ColorRGBW(0, 0, 0, 0.2), ColorRGBW(0, 0, 0, 1)),
|
self.protocol: MusicMouseProtocol = protocol
|
||||||
}
|
self.button_led_brightness = None
|
||||||
|
|
||||||
playlists = {
|
def mouse_led_effect(self, effect_cfg):
|
||||||
fig: audio_player.create_playlist(sorted(glob(os.path.join(MUSIC_FOLDER, fig, "*.mp3"))))
|
self.current_mouse_led_effect = effect_cfg
|
||||||
for fig in rfid_token_map.values()
|
self.protocol.mouse_led_effect(effect_cfg)
|
||||||
}
|
|
||||||
|
|
||||||
mouse_leds = {
|
def led_ring_effect(self, effect_cfg):
|
||||||
TouchButton.RIGHT_FOOT: (0, 6),
|
self.current_led_ring_effect = effect_cfg
|
||||||
TouchButton.LEFT_FOOT: (6, 6 + 6),
|
self.protocol.led_ring_effect(effect_cfg)
|
||||||
TouchButton.LEFT_EAR: (6 + 6, 6 + 6 + 16),
|
|
||||||
TouchButton.RIGHT_EAR: (6 + 6 + 16, 6 + 6 + 16 + 17),
|
def button_leds(self, brightness):
|
||||||
}
|
assert 0 <= brightness <= 1
|
||||||
|
self.protocol.button_background_led_prev(brightness)
|
||||||
|
self.protocol.button_background_led_next(brightness)
|
||||||
|
self.button_led_brightness = brightness
|
||||||
|
|
||||||
|
def reset(self):
|
||||||
|
self.mouse_led_effect(EffectStaticConfig(OFF_COLOR))
|
||||||
|
self.led_ring_effect(EffectStaticConfig(OFF_COLOR))
|
||||||
|
|
||||||
|
def figure_placed(self, figure_state):
|
||||||
|
self.last_figure = self.current_figure
|
||||||
|
self.current_figure = figure_state
|
||||||
|
|
||||||
|
def figure_removed(self):
|
||||||
|
self.last_figure = self.current_figure
|
||||||
|
|
||||||
|
|
||||||
def run_off_animation(protocol):
|
class Controller:
|
||||||
ring_eff = EffectReverseSwipe()
|
def __init__(self, protocol, cfg):
|
||||||
mouse_eff = EffectReverseSwipe()
|
self.cfg = cfg
|
||||||
mouse_eff.startPosition = 6 / 45 * 360
|
self.audio_player = AudioPlayer(cfg["general"]["alsa_device"])
|
||||||
protocol.led_ring_effect(ring_eff)
|
self.mmstate = MusicMouseState(protocol)
|
||||||
protocol.mouse_led_effect(mouse_eff)
|
|
||||||
protocol.button_background_led_prev(0)
|
|
||||||
protocol.button_background_led_next(0)
|
|
||||||
|
|
||||||
|
protocol.register_message_callback(self.on_firmware_msg)
|
||||||
|
|
||||||
def on_music_end_callback(protocol):
|
self.audio_player.on_playlist_end_callback = self._run_off_animation
|
||||||
run_off_animation(protocol)
|
self.playlists = {
|
||||||
|
fig: self.audio_player.create_playlist(fig_cfg['media_files'])
|
||||||
|
for fig, fig_cfg in cfg['figures'].items()
|
||||||
|
}
|
||||||
|
self._rfid_to_figure_name = {
|
||||||
|
bytes.fromhex(figure_cfg["id"]): figure_name
|
||||||
|
for figure_name, figure_cfg in cfg["figures"].items()
|
||||||
|
}
|
||||||
|
|
||||||
|
def handle_rfid_event(self, tagid):
|
||||||
|
if tagid == bytes.fromhex("0000000000"):
|
||||||
|
if self.audio_player.is_playing():
|
||||||
|
self._run_off_animation()
|
||||||
|
self.audio_player.pause()
|
||||||
|
self.mmstate.figure_removed()
|
||||||
|
elif tagid in self._rfid_to_figure_name:
|
||||||
|
figure = self._rfid_to_figure_name[tagid]
|
||||||
|
primary_color, secondary_color, *rest = self.cfg["figures"]["colors"]
|
||||||
|
self._start_animation(primary_color, secondary_color)
|
||||||
|
self.mmstate.button_leds(self.cfg["general"].get("button_leds_brightness", 0.5))
|
||||||
|
|
||||||
def on_rfid(protocol, tagid):
|
if figure in self.playlists:
|
||||||
global current_figure, last_figure
|
self.audio_player.set_playlist(self.playlists[figure])
|
||||||
|
if self.mmstate.last_figure == figure:
|
||||||
|
self.audio_player.play()
|
||||||
|
else:
|
||||||
|
self.audio_player.play_from_start()
|
||||||
|
|
||||||
if tagid == bytes.fromhex("0000000000"):
|
self.mmstate.figure_placed(figure)
|
||||||
# Off
|
else:
|
||||||
if audio_player.is_playing():
|
warnings.warn(f"Unknown figure/tag with id {tagid}")
|
||||||
run_off_animation(protocol)
|
|
||||||
audio_player.pause()
|
def on_firmware_msg(self, _, message):
|
||||||
#else:
|
print("FW msg:", message)
|
||||||
# protocol.led_ring_effect(EffectStaticConfig(ColorRGBW(0, 0, 0, 0)))
|
if isinstance(message, RfidTokenRead):
|
||||||
# protocol.mouse_led_effect(EffectStaticConfig(ColorRGBW(0, 0, 0, 0)))
|
self.handle_rfid_event(message.id)
|
||||||
last_figure = current_figure
|
elif isinstance(message, RotaryEncoderEvent):
|
||||||
else:
|
volume_increment = self.cfg["general"].get("volume_increment", 2)
|
||||||
figure = rfid_token_map[tagid]
|
if message.direction == 2:
|
||||||
current_figure = figure
|
self.audio_player.change_volume(volume_increment)
|
||||||
ccfg = color_cfg[figure]
|
elif message.direction == 1:
|
||||||
|
self.audio_player.change_volume(-volume_increment)
|
||||||
|
elif isinstance(message, ButtonEvent):
|
||||||
|
if message.button == "left" and message.event == "pressed":
|
||||||
|
self.audio_player.previous()
|
||||||
|
elif message.button == "right" and message.event == "pressed":
|
||||||
|
self.audio_player.next()
|
||||||
|
elif False and isinstance(message, TouchButtonPress):
|
||||||
|
if current_figure:
|
||||||
|
ccfg = color_cfg[current_figure]
|
||||||
|
protocol.mouse_led_effect(
|
||||||
|
EffectStaticConfig(ccfg.accent, *mouse_leds[message.touch_button]))
|
||||||
|
#if message.touch_button == TouchButton.RIGHT_FOOT:
|
||||||
|
# asyncio.create_task(
|
||||||
|
# hass.call_service("switch", "toggle", {"entity_id": "switch.tasmota06"}))
|
||||||
|
# asyncio.create_task(
|
||||||
|
# hass.call_service("light", "turn_on", {"entity_id": "light.arbeitszimmer_fluter"}))
|
||||||
|
#elif message.touch_button == TouchButton.LEFT_FOOT:
|
||||||
|
# asyncio.create_task(
|
||||||
|
# hass.call_service("light", "turn_off", {"entity_id": "light.arbeitszimmer_fluter"}))
|
||||||
|
|
||||||
|
elif False and isinstance(message, TouchButtonRelease):
|
||||||
|
if current_figure:
|
||||||
|
ccfg = color_cfg[current_figure]
|
||||||
|
eff_change = EffectRandomTwoColorInterpolationConfig()
|
||||||
|
eff_static = EffectStaticConfig(ColorRGBW(0, 0, 0, 0),
|
||||||
|
*mouse_leds[message.touch_button])
|
||||||
|
if audio_player.is_playing():
|
||||||
|
eff_static.color = ccfg.primary
|
||||||
|
protocol.mouse_led_effect(eff_static)
|
||||||
|
|
||||||
|
if audio_player.is_playing():
|
||||||
|
eff_change.color1 = ccfg.primary
|
||||||
|
eff_change.color2 = ccfg.secondary
|
||||||
|
eff_change.start_with_existing = True
|
||||||
|
protocol.mouse_led_effect(eff_change)
|
||||||
|
|
||||||
|
def _start_animation(self, primary_color, secondary_color):
|
||||||
ring_eff = EffectSwipeAndChange()
|
ring_eff = EffectSwipeAndChange()
|
||||||
ring_eff.swipe.primary_color = ccfg.primary
|
ring_eff.swipe.primary_color = primary_color
|
||||||
ring_eff.swipe.secondary_color = ccfg.secondary
|
ring_eff.swipe.secondary_color = secondary_color
|
||||||
ring_eff.swipe.swipe_speed = 180
|
ring_eff.swipe.swipe_speed = 180
|
||||||
ring_eff.change.color1 = ccfg.primary
|
ring_eff.change.color1 = primary_color
|
||||||
ring_eff.change.color2 = ccfg.secondary
|
ring_eff.change.color2 = secondary_color
|
||||||
protocol.led_ring_effect(ring_eff)
|
self.mmstate.led_ring_effect(ring_eff)
|
||||||
|
|
||||||
mouse_eff = EffectStaticConfig(ccfg.background)
|
|
||||||
mouse_eff = deepcopy(ring_eff)
|
mouse_eff = deepcopy(ring_eff)
|
||||||
mouse_eff.swipe.start_position = 6 / 45 * 360
|
mouse_eff.swipe.start_position = 6 / 45 * 360
|
||||||
mouse_eff.swipe.bell_curve_width_in_leds = 16
|
mouse_eff.swipe.bell_curve_width_in_leds = 16
|
||||||
mouse_eff.swipe.swipe_speed = 180
|
mouse_eff.swipe.swipe_speed = 180
|
||||||
protocol.mouse_led_effect(mouse_eff)
|
self.mmstate.mouse_led_effect(mouse_eff)
|
||||||
|
|
||||||
protocol.button_background_led_prev(0.3)
|
def _run_off_animation(self):
|
||||||
protocol.button_background_led_next(0.3)
|
ring_eff = EffectReverseSwipe()
|
||||||
|
self.mmstate.led_ring_effect(ring_eff)
|
||||||
|
|
||||||
if figure in playlists:
|
mouse_eff = EffectReverseSwipe()
|
||||||
audio_player.set_playlist(playlists[figure])
|
mouse_eff.startPosition = 6 / 45 * 360
|
||||||
if last_figure == current_figure:
|
self.mmstate.mouse_led_effect(mouse_eff)
|
||||||
audio_player.play()
|
|
||||||
else:
|
self.mmstate.button_leds(0)
|
||||||
audio_player.play_from_start()
|
|
||||||
|
|
||||||
|
|
||||||
def on_firmware_msg(protocol: MusicMouseProtocol, message):
|
def main(config_path):
|
||||||
print("FW msg:", message)
|
cfg = load_config(config_path)
|
||||||
if isinstance(message, RfidTokenRead):
|
loop = asyncio.get_event_loop()
|
||||||
on_rfid(protocol, message.id)
|
coro = serial_asyncio.create_serial_connection(loop,
|
||||||
elif isinstance(message, RotaryEncoderEvent):
|
MusicMouseProtocol,
|
||||||
if audio_player.is_playing():
|
cfg["general"]["serial_port"],
|
||||||
if message.direction == 2:
|
baudrate=115200)
|
||||||
audio_player.change_volume(2)
|
transport, protocol = loop.run_until_complete(coro)
|
||||||
elif message.direction == 1:
|
controller = Controller(protocol, cfg)
|
||||||
audio_player.change_volume(-2)
|
return controller, loop
|
||||||
elif isinstance(message, ButtonEvent):
|
|
||||||
if message.button == "left" and message.event == "pressed":
|
|
||||||
audio_player.previous()
|
|
||||||
elif message.button == "right" and message.event == "pressed":
|
|
||||||
audio_player.next()
|
|
||||||
elif False and isinstance(message, TouchButtonPress):
|
|
||||||
if current_figure:
|
|
||||||
ccfg = color_cfg[current_figure]
|
|
||||||
protocol.mouse_led_effect(
|
|
||||||
EffectStaticConfig(ccfg.accent, *mouse_leds[message.touch_button]))
|
|
||||||
#if message.touch_button == TouchButton.RIGHT_FOOT:
|
|
||||||
# asyncio.create_task(
|
|
||||||
# hass.call_service("switch", "toggle", {"entity_id": "switch.tasmota06"}))
|
|
||||||
# asyncio.create_task(
|
|
||||||
# hass.call_service("light", "turn_on", {"entity_id": "light.arbeitszimmer_fluter"}))
|
|
||||||
#elif message.touch_button == TouchButton.LEFT_FOOT:
|
|
||||||
# asyncio.create_task(
|
|
||||||
# hass.call_service("light", "turn_off", {"entity_id": "light.arbeitszimmer_fluter"}))
|
|
||||||
|
|
||||||
elif False and isinstance(message, TouchButtonRelease):
|
|
||||||
if current_figure:
|
|
||||||
ccfg = color_cfg[current_figure]
|
|
||||||
eff_change = EffectRandomTwoColorInterpolationConfig()
|
|
||||||
eff_static = EffectStaticConfig(ColorRGBW(0, 0, 0, 0), *mouse_leds[message.touch_button])
|
|
||||||
if audio_player.is_playing():
|
|
||||||
eff_static.color = ccfg.primary
|
|
||||||
protocol.mouse_led_effect(eff_static)
|
|
||||||
|
|
||||||
if audio_player.is_playing():
|
|
||||||
eff_change.color1 = ccfg.primary
|
|
||||||
eff_change.color2 = ccfg.secondary
|
|
||||||
eff_change.start_with_existing = True
|
|
||||||
protocol.mouse_led_effect(eff_change)
|
|
||||||
|
|
||||||
|
|
||||||
loop = asyncio.get_event_loop()
|
if __name__ == "__main__":
|
||||||
hass = HomeAssistantClient(HASS_URL, HASS_TOKEN, loop)
|
if len(sys.argv) == 2:
|
||||||
coro = serial_asyncio.create_serial_connection(loop,
|
controller, loop = main(config_path=sys.argv[1])
|
||||||
MusicMouseProtocol,
|
loop.run_forever()
|
||||||
'/dev/ttyUSB0',
|
loop.close()
|
||||||
baudrate=115200)
|
else:
|
||||||
transport, protocol = loop.run_until_complete(coro)
|
print("Error: run with config file path as first argument")
|
||||||
protocol.register_message_callback(on_firmware_msg)
|
|
||||||
|
|
||||||
audio_player.on_playlist_end_callback = lambda: on_music_end_callback(protocol)
|
|
||||||
|
|
||||||
loop.create_task(hass.connect())
|
|
||||||
loop.run_forever()
|
|
||||||
loop.close()
|
|
||||||
|
|
|
@ -69,8 +69,9 @@ all_events = (
|
||||||
|
|
||||||
|
|
||||||
class AudioPlayer:
|
class AudioPlayer:
|
||||||
def __init__(self):
|
def __init__(self, alsa_device=None):
|
||||||
self.instance = vlc.Instance()
|
params = ["-A", "alsa", "--alsa-audio-device", alsa_device] if alsa_device else []
|
||||||
|
self.instance = vlc.Instance(*params)
|
||||||
self.media_list_player = self.instance.media_list_player_new()
|
self.media_list_player = self.instance.media_list_player_new()
|
||||||
self.media_player = self.media_list_player.get_media_player()
|
self.media_player = self.media_list_player.get_media_player()
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue