Skip to content

Commit

Permalink
Add global option to skip confirmation when using Telnet
Browse files Browse the repository at this point in the history
Fixes #321
  • Loading branch information
henrikwidlund committed Jan 25, 2025
1 parent 7229456 commit af3f751
Showing 1 changed file with 6 additions and 4 deletions.
10 changes: 6 additions & 4 deletions denonavr/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -481,6 +481,7 @@ class DenonAVRTelnetApi:
_reconnect_task: asyncio.Task = attr.ib(default=None)
_monitor_handle: asyncio.TimerHandle = attr.ib(default=None)
_protocol: DenonAVRTelnetProtocol = attr.ib(default=None)
_skip_confirmation: bool = attr.lib(default=False)
_telnet_event_map: Dict[str, List] = attr.ib(
default=attr.Factory(telnet_event_map_factory)
)
Expand Down Expand Up @@ -812,8 +813,9 @@ async def _async_send_command(
self, command: str, skip_confirmation: bool = False
) -> None:
"""Send one telnet command to the receiver."""
use_skip_confirmation = skip_confirmation if skip_confirmation is not None else self._skip_confirmation
async with self._send_lock:
if not skip_confirmation:
if not use_skip_confirmation:
self._send_confirmation_command = command
self._send_confirmation_event.clear()
if not self.connected or not self.healthy:
Expand All @@ -822,7 +824,7 @@ async def _async_send_command(
f"{self.connected}, Connection healthy: {self.healthy}"
)
self._protocol.write(f"{command}\r")
if not skip_confirmation:
if not use_skip_confirmation:
try:
await asyncio.wait_for(
self._send_confirmation_event.wait(),
Expand All @@ -836,13 +838,13 @@ async def _async_send_command(
self._send_confirmation_command = ""

async def async_send_commands(
self, *commands: str, skip_confirmation: bool = False
self, *commands: str, skip_confirmation: Optional[bool] = None
) -> None:
"""Send telnet commands to the receiver."""
for command in commands:
await self._async_send_command(command, skip_confirmation=skip_confirmation)

def send_commands(self, *commands: str, skip_confirmation: bool = False) -> None:
def send_commands(self, *commands: str, skip_confirmation: Optional[bool] = None) -> None:
"""Send telnet commands to the receiver."""
task = asyncio.create_task(
self.async_send_commands(*commands, skip_confirmation=skip_confirmation)
Expand Down

0 comments on commit af3f751

Please sign in to comment.