Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

New simple commands: dim display, power toggle, trigger 1 & 2 #40

Merged
merged 2 commits into from
Jun 14, 2024
Merged

New simple commands: dim display, power toggle, trigger 1 & 2 #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e9eb781
refactor: command handler, simple commands enum
zehnm Jun 14, 2024
dacf304
feat: power toggle, dimmer, trigger commands
zehnm Jun 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 26 additions & 74 deletions intg-denonavr/avr.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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")

@async_handle_denonlib_errors
async def output_monitor_1(self) -> ucapi.StatusCodes:
"""Send cursor down 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 cursor down 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 cursor down 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")
return await self.send_command("MNOPT")

@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:
Expand All @@ -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:
Expand Down
116 changes: 66 additions & 50 deletions intg-denonavr/media_player.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,22 @@
}


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):
"""Representation of a Denon Media Player entity."""

Expand All @@ -43,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,
Expand Down Expand Up @@ -77,7 +94,7 @@ 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 = [*SimpleCommandMappings]

options = {Options.SIMPLE_COMMANDS: self.simple_commands}

Expand All @@ -90,7 +107,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.

Expand All @@ -106,54 +123,53 @@ 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.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:
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 cmd if cmd in SimpleCommandMappings:
res = await self._receiver.send_command(SimpleCommandMappings[cmd])
case _:
return StatusCodes.NOT_IMPLEMENTED

return res

Expand Down
Loading