From e9eb781614b706d323f718759d3e72e720889b14 Mon Sep 17 00:00:00 2001 From: Markus Zehnder Date: Fri, 14 Jun 2024 15:13:48 +0200 Subject: [PATCH 1/2] refactor: command handler, simple commands enum Use match-case in the command handler and an enum for all simple commands. --- intg-denonavr/avr.py | 6 +- intg-denonavr/media_player.py | 117 +++++++++++++++++++--------------- 2 files changed, 70 insertions(+), 53 deletions(-) diff --git a/intg-denonavr/avr.py b/intg-denonavr/avr.py index 28b2c1c..470c8bd 100644 --- a/intg-denonavr/avr.py +++ b/intg-denonavr/avr.py @@ -797,7 +797,7 @@ async def options(self) -> ucapi.StatusCodes: @async_handle_denonlib_errors async def output_monitor_1(self) -> ucapi.StatusCodes: - """Send cursor down command to AVR.""" + """Send set HDMI monitor out 1 command to AVR.""" if self._use_telnet: await self._receiver.async_send_telnet_commands("VSMONI1") else: @@ -805,7 +805,7 @@ async def output_monitor_1(self) -> ucapi.StatusCodes: @async_handle_denonlib_errors async def output_monitor_2(self) -> ucapi.StatusCodes: - """Send cursor down command to AVR.""" + """Send set HDMI monitor out 2 command to AVR.""" if self._use_telnet: await self._receiver.async_send_telnet_commands("VSMONI2") else: @@ -813,7 +813,7 @@ async def output_monitor_2(self) -> ucapi.StatusCodes: @async_handle_denonlib_errors async def output_monitor_auto(self) -> ucapi.StatusCodes: - """Send cursor down command to AVR.""" + """Send set HDMI monitor automatic detection command to AVR.""" if self._use_telnet: await self._receiver.async_send_telnet_commands("VSMONIAUTO") else: diff --git a/intg-denonavr/media_player.py b/intg-denonavr/media_player.py index c76d849..6a08001 100644 --- a/intg-denonavr/media_player.py +++ b/intg-denonavr/media_player.py @@ -6,6 +6,7 @@ """ import logging +from enum import Enum from typing import Any import avr @@ -33,6 +34,17 @@ } +class SimpleCommands(str, Enum): + """Additional simple commands of the Denon AVR not covered by media-player features.""" + + OUTPUT_1 = "OUTPUT_1" + """Set HDMI monitor out 1.""" + OUTPUT_2 = "OUTPUT_2" + """Set HDMI monitor out 2.""" + OUTPUT_AUTO = "OUTPUT_AUTO" + """Set HDMI monitor automatic detection.""" + + class DenonMediaPlayer(MediaPlayer): """Representation of a Denon Media Player entity.""" @@ -77,7 +89,11 @@ def __init__(self, device: AvrDevice, receiver: avr.DenonDevice): attributes[Attributes.SOUND_MODE] = "" attributes[Attributes.SOUND_MODE_LIST] = [] - self.simple_commands = ["OUTPUT_1", "OUTPUT_2", "OUTPUT_AUTO"] + self.simple_commands = [ + SimpleCommands.OUTPUT_1.value, + SimpleCommands.OUTPUT_2.value, + SimpleCommands.OUTPUT_AUTO.value, + ] options = {Options.SIMPLE_COMMANDS: self.simple_commands} @@ -90,7 +106,7 @@ def __init__(self, device: AvrDevice, receiver: avr.DenonDevice): options=options, ) - async def command(self, cmd_id: str, params: dict[str, Any] | None = None) -> StatusCodes: # pylint: disable=R0915 + async def command(self, cmd_id: str, params: dict[str, Any] | None = None) -> StatusCodes: """ Media-player entity command handler. @@ -106,54 +122,55 @@ async def command(self, cmd_id: str, params: dict[str, Any] | None = None) -> St _LOG.warning("No AVR instance for entity: %s", self.id) return StatusCodes.SERVICE_UNAVAILABLE - if cmd_id == Commands.PLAY_PAUSE: - res = await self._receiver.play_pause() - elif cmd_id == Commands.NEXT: - res = await self._receiver.next() - elif cmd_id == Commands.PREVIOUS: - res = await self._receiver.previous() - elif cmd_id == Commands.VOLUME: - res = await self._receiver.set_volume_level(params.get("volume")) - elif cmd_id == Commands.VOLUME_UP: - res = await self._receiver.volume_up() - elif cmd_id == Commands.VOLUME_DOWN: - res = await self._receiver.volume_down() - elif cmd_id == Commands.MUTE_TOGGLE: - res = await self._receiver.mute(not self.attributes[Attributes.MUTED]) - elif cmd_id == Commands.ON: - res = await self._receiver.power_on() - elif cmd_id == Commands.OFF: - res = await self._receiver.power_off() - elif cmd_id == Commands.SELECT_SOURCE: - res = await self._receiver.select_source(params.get("source")) - elif cmd_id == Commands.SELECT_SOUND_MODE: - res = await self._receiver.select_sound_mode(params.get("mode")) - elif cmd_id == Commands.CURSOR_UP: - res = await self._receiver.cursor_up() - elif cmd_id == Commands.CURSOR_DOWN: - res = await self._receiver.cursor_down() - elif cmd_id == Commands.CURSOR_LEFT: - res = await self._receiver.cursor_left() - elif cmd_id == Commands.CURSOR_RIGHT: - res = await self._receiver.cursor_right() - elif cmd_id == Commands.CURSOR_ENTER: - res = await self._receiver.cursor_enter() - elif cmd_id == Commands.BACK: - res = await self._receiver.back() - elif cmd_id == Commands.MENU: - res = await self._receiver.setup() - elif cmd_id == Commands.CONTEXT_MENU: - res = await self._receiver.options() - elif cmd_id == Commands.INFO: - res = await self._receiver.info() - elif cmd_id == "OUTPUT_1": - res = await self._receiver.output_monitor_1() - elif cmd_id == "OUTPUT_2": - res = await self._receiver.output_monitor_2() - elif cmd_id == "OUTPUT_AUTO": - res = await self._receiver.output_monitor_auto() - else: - return StatusCodes.NOT_IMPLEMENTED + match cmd_id: + case Commands.PLAY_PAUSE: + res = await self._receiver.play_pause() + case Commands.NEXT: + res = await self._receiver.next() + case Commands.PREVIOUS: + res = await self._receiver.previous() + case Commands.VOLUME: + res = await self._receiver.set_volume_level(params.get("volume")) + case Commands.VOLUME_UP: + res = await self._receiver.volume_up() + case Commands.VOLUME_DOWN: + res = await self._receiver.volume_down() + case Commands.MUTE_TOGGLE: + res = await self._receiver.mute(not self.attributes[Attributes.MUTED]) + case Commands.ON: + res = await self._receiver.power_on() + case Commands.OFF: + res = await self._receiver.power_off() + case Commands.SELECT_SOURCE: + res = await self._receiver.select_source(params.get("source")) + case Commands.SELECT_SOUND_MODE: + res = await self._receiver.select_sound_mode(params.get("mode")) + case Commands.CURSOR_UP: + res = await self._receiver.cursor_up() + case Commands.CURSOR_DOWN: + res = await self._receiver.cursor_down() + case Commands.CURSOR_LEFT: + res = await self._receiver.cursor_left() + case Commands.CURSOR_RIGHT: + res = await self._receiver.cursor_right() + case Commands.CURSOR_ENTER: + res = await self._receiver.cursor_enter() + case Commands.BACK: + res = await self._receiver.back() + case Commands.MENU: + res = await self._receiver.setup() + case Commands.CONTEXT_MENU: + res = await self._receiver.options() + case Commands.INFO: + res = await self._receiver.info() + case SimpleCommands.OUTPUT_1: + res = await self._receiver.output_monitor_1() + case SimpleCommands.OUTPUT_2: + res = await self._receiver.output_monitor_2() + case SimpleCommands.OUTPUT_AUTO: + res = await self._receiver.output_monitor_auto() + case _: + return StatusCodes.NOT_IMPLEMENTED return res From dacf304509256e06e347183f9db06f93f76df239 Mon Sep 17 00:00:00 2001 From: Markus Zehnder Date: Fri, 14 Jun 2024 18:19:08 +0200 Subject: [PATCH 2/2] feat: power toggle, dimmer, trigger commands - implement power toggle in driver, based on current power state - support dimmer commands - support trigger 1 & 2 on / off commands --- intg-denonavr/avr.py | 100 +++++++++------------------------- intg-denonavr/media_player.py | 41 +++++++------- 2 files changed, 46 insertions(+), 95 deletions(-) diff --git a/intg-denonavr/avr.py b/intg-denonavr/avr.py index 470c8bd..7123ae1 100644 --- a/intg-denonavr/avr.py +++ b/intg-denonavr/avr.py @@ -652,6 +652,13 @@ async def power_off(self) -> ucapi.StatusCodes: if not self._use_telnet: self._set_expected_state(States.OFF) + @async_handle_denonlib_errors + async def power_toggle(self) -> ucapi.StatusCodes: + """Send power-on or -off command to AVR based on current power state.""" + if self._receiver.power is not None and self._receiver.power == "ON": + await self.power_off() + await self.power_on() + @async_handle_denonlib_errors async def set_volume_level(self, volume: float | None) -> ucapi.StatusCodes: """Set volume level, range 0..100.""" @@ -732,119 +739,55 @@ async def select_sound_mode(self, sound_mode: str | None) -> ucapi.StatusCodes: return ucapi.StatusCodes.BAD_REQUEST await self._receiver.async_set_sound_mode(sound_mode) - @async_handle_denonlib_errors async def cursor_up(self) -> ucapi.StatusCodes: """Send cursor up command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNCUP") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNCUP") + return await self.send_command("MNCUP") - @async_handle_denonlib_errors async def cursor_down(self) -> ucapi.StatusCodes: """Send cursor down command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNCDN") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNCDN") + return await self.send_command("MNCDN") - @async_handle_denonlib_errors async def cursor_left(self) -> ucapi.StatusCodes: """Send cursor left command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNCLT") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNCLT") + return await self.send_command("MNCLT") - @async_handle_denonlib_errors async def cursor_right(self) -> ucapi.StatusCodes: """Send cursor right command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNCRT") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNCRT") + return await self.send_command("MNCRT") - @async_handle_denonlib_errors async def cursor_enter(self) -> ucapi.StatusCodes: """Send cursor enter command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNENT") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNENT") + return await self.send_command("MNENT") - @async_handle_denonlib_errors async def info(self) -> ucapi.StatusCodes: """Send info OSD command command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNINF") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNINF") + return await self.send_command("MNINF") - @async_handle_denonlib_errors async def options(self) -> ucapi.StatusCodes: """Send options menu command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNOPT") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNOPT") + return await self.send_command("MNOPT") - @async_handle_denonlib_errors - async def output_monitor_1(self) -> ucapi.StatusCodes: - """Send set HDMI monitor out 1 command to AVR.""" - if self._use_telnet: - await self._receiver.async_send_telnet_commands("VSMONI1") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?VSMONI1") - - @async_handle_denonlib_errors - async def output_monitor_2(self) -> ucapi.StatusCodes: - """Send set HDMI monitor out 2 command to AVR.""" - if self._use_telnet: - await self._receiver.async_send_telnet_commands("VSMONI2") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?VSMONI2") - - @async_handle_denonlib_errors - async def output_monitor_auto(self) -> ucapi.StatusCodes: - """Send set HDMI monitor automatic detection command to AVR.""" - if self._use_telnet: - await self._receiver.async_send_telnet_commands("VSMONIAUTO") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?VSMONIAUTO") - - @async_handle_denonlib_errors async def back(self) -> ucapi.StatusCodes: """Send back command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNRTN") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNRTN") + return await self.send_command("MNRTN") - @async_handle_denonlib_errors async def setup_open(self) -> ucapi.StatusCodes: """Send open setup menu command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNMEN ON") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNMEN%20ON") + return await self.send_command("MNMEN ON") - @async_handle_denonlib_errors async def setup_close(self) -> ucapi.StatusCodes: """Send close menu command to AVR.""" # TODO : to be updated when PR will be released https://github.com/ol-iver/denonavr/pull/290 - if self._use_telnet: - await self._receiver.async_send_telnet_commands("MNMEN OFF") - else: - await self._receiver.async_get_command(AVR_COMMAND_URL + "?MNMEN%20OFF") + return await self.send_command("MNMEN OFF") @async_handle_denonlib_errors async def setup(self) -> ucapi.StatusCodes: @@ -857,6 +800,15 @@ async def setup(self) -> ucapi.StatusCodes: else: await self.setup_open() + @async_handle_denonlib_errors + async def send_command(self, cmd: str) -> ucapi.StatusCodes: + """Send a command to the AVR.""" + if self._use_telnet: + await self._receiver.async_send_telnet_commands(cmd) + else: + url = AVR_COMMAND_URL + "?" + cmd.replace(" ", "%20") + await self._receiver.async_get_command(url) + def _increase_expected_volume(self): """Without telnet, increase expected volume and send update event.""" if not self._use_telnet or self._expected_volume is None: diff --git a/intg-denonavr/media_player.py b/intg-denonavr/media_player.py index 6a08001..14b3208 100644 --- a/intg-denonavr/media_player.py +++ b/intg-denonavr/media_player.py @@ -6,7 +6,6 @@ """ import logging -from enum import Enum from typing import Any import avr @@ -34,15 +33,20 @@ } -class SimpleCommands(str, Enum): - """Additional simple commands of the Denon AVR not covered by media-player features.""" - - OUTPUT_1 = "OUTPUT_1" - """Set HDMI monitor out 1.""" - OUTPUT_2 = "OUTPUT_2" - """Set HDMI monitor out 2.""" - OUTPUT_AUTO = "OUTPUT_AUTO" - """Set HDMI monitor automatic detection.""" +SimpleCommandMappings = { + "OUTPUT_1": "VSMONI1", + "OUTPUT_2": "VSMONI2", + "OUTPUT_AUTO": "VSMONIAUTO", + "DIMMER_TOGGLE": "DIM SEL", + "DIMMER_BRIGHT": "DIM BRI", + "DIMMER_DIM": "DIM DIM", + "DIMMER_DARK": "DIM DAR", + "DIMMER_OFF": "DIM OFF", + "TRIGGER1_ON": "TR1 ON", + "TRIGGER1_OFF": "TR1 OFF", + "TRIGGER2_ON": "TR2 ON", + "TRIGGER2_OFF": "TR2 OFF", +} class DenonMediaPlayer(MediaPlayer): @@ -55,6 +59,7 @@ def __init__(self, device: AvrDevice, receiver: avr.DenonDevice): entity_id = create_entity_id(receiver.id, EntityTypes.MEDIA_PLAYER) features = [ Features.ON_OFF, + Features.TOGGLE, Features.VOLUME, Features.VOLUME_UP_DOWN, Features.MUTE_TOGGLE, @@ -89,11 +94,7 @@ def __init__(self, device: AvrDevice, receiver: avr.DenonDevice): attributes[Attributes.SOUND_MODE] = "" attributes[Attributes.SOUND_MODE_LIST] = [] - self.simple_commands = [ - SimpleCommands.OUTPUT_1.value, - SimpleCommands.OUTPUT_2.value, - SimpleCommands.OUTPUT_AUTO.value, - ] + self.simple_commands = [*SimpleCommandMappings] options = {Options.SIMPLE_COMMANDS: self.simple_commands} @@ -141,6 +142,8 @@ async def command(self, cmd_id: str, params: dict[str, Any] | None = None) -> St res = await self._receiver.power_on() case Commands.OFF: res = await self._receiver.power_off() + case Commands.TOGGLE: + res = await self._receiver.power_toggle() case Commands.SELECT_SOURCE: res = await self._receiver.select_source(params.get("source")) case Commands.SELECT_SOUND_MODE: @@ -163,12 +166,8 @@ async def command(self, cmd_id: str, params: dict[str, Any] | None = None) -> St res = await self._receiver.options() case Commands.INFO: res = await self._receiver.info() - case SimpleCommands.OUTPUT_1: - res = await self._receiver.output_monitor_1() - case SimpleCommands.OUTPUT_2: - res = await self._receiver.output_monitor_2() - case SimpleCommands.OUTPUT_AUTO: - res = await self._receiver.output_monitor_auto() + case cmd if cmd in SimpleCommandMappings: + res = await self._receiver.send_command(SimpleCommandMappings[cmd]) case _: return StatusCodes.NOT_IMPLEMENTED