[WIP] Events, LED

[wmantly]

This is a work in progress to add features for LED and button(s) on the 2mic pi hat. This PR will also include some other house keeping and general quality of life concerns

Completed

• Added pub/sud event bus to allow the code to be decoupled
• Added LED file and events. The LED lights now:
  1. Blink green 3 times when the app service connects to HA
  2. Pulses blue when wake word is activated until VAD end
  3. Pulses yellow while awaiting a response from the server
  4. Pulses green while player response from the server
  5. Is off at all other times

Todo

• Abstract the events_led.py code so it will also work with the 4mic. The code can be abstracted to work with any LED/light via pub/sub.
• Allow a config file to also be used instead of just command line arguments. I personally see this as a QoL issue.
• Write a shell script to set up/start the service.
• Add support for button(s)

I would love comments on this PR, and also some guidance/links to working with the ESPhome package and API.

[hkfuertes]

Completely outsider that just landed on the project... I think the bus idea is great and can be used for a lot of things! I see there is another pr with thinking sound, etc that too can go through the bus... It would be awesome if this bus can be accessed externally... kind of the way rhasspy does it... but maybe not, meaning exposing a server to where several clients can connect and read events from the bus... so that we can have a 2mics led client, a 4mics led client but also a thinking_sound client... etc... but this channel can also be bidirectional, so a push_button client can also be implemented to trigger the listening...

Don't know... my two cents... for now I'm focussed on build the bare project onto my buildroot image :)

[Teagan42]

[Suggestion] Configurable handler metadata:

Suggested change

	def subscribe(func: Callable) -> Callable:
	"""Decorator to mark a method for event bus subscription."""
	func._event_bus_subscribe = True
	return func
	@dataclass
	class EventBusMetaData:
	priority: int = 0
	# Any other info related to the handler
	def subscribe(priority: int=0, **kwargs):
	"""Decorator to mark a method for event bus subscription."""
	def wrapper(func: Callable) -> Callable:
	"""Function wrapper."""
	func._event_bus_subscribe = True
	func._event_bus_meta = EventBusMetaData(
	priority=priority,
	...
	)
	return func
	return wrapper

[Teagan42]

[Nitpick] An event bus is based off events - not topics, it would scale far more easily to have event handlers react to VoiceAssistantEvents

[Teagan42]

[Suggestion] Based on below suggestion:

Suggested change

	def publish(self, topic: str, data: [dict, None]) -> None:
	"""
	Publishes an event to all subscribed listeners.
	"""
	# _LOGGER.debug(f'EventBus publish {topic}')
	data['__topic'] = topic
	listeners = self.topics.get(topic, [])
	for listener in listeners:
	listener(data)
	async def publish(self, topic: str, data: [dict, None]) -> None:
	"""
	Publishes an event to all subscribed listeners.
	"""
	# _LOGGER.debug(f'EventBus publish {topic}')
	data['__topic'] = topic
	listeners = self.topics.get(topic, [])
	for listener in sorted(listeners, key=lambda l: l._event_bus_priority):
	if inspect.iscoroutine(listener):
	await listener(data)
	else:
	listener(data)

When publishing events, use asyncio.create_task to publish the event.

[florian-asche]

Can you specify if this is for the v1 or the v2 of the 2MicHat?
And i hope we can get some sort of configuration parameter for possible boards here. I have the v1 and the ReSpeaker Lite here on my desk.

[florian-asche]

We have discussions now. It would be great, if you can add one for the LED part and hope you give some input from what you already did, maybe future ideas etc.